Refine WebUI commands and data skills

src/agent_tools.py

@@ -5,7 +5,9 @@ import json
 import os
 import re
 import selectors
+import shutil
 import subprocess
+import sys
 import time
 import urllib.error
 import urllib.parse
@@ -333,9 +335,55 @@ def default_tool_registry() -> dict[str, AgentTool]:
             },
             handler=_grep_search,
         ),
+        AgentTool(
+            name='python_exec',
+            description=(
+                '优先用本工具执行小型 Python 代码或项目内 Python 脚本，用于结构化文件分析、'
+                'JSON/JSONL 处理、批量校验、数据抽样和快速计算。默认使用项目 .venv/bin/python；'
+                '不要用 bash 运行 python/python3/.venv/bin/python。不要用本工具安装依赖。'
+                '缺包时应向用户确认后再处理依赖。'
+            ),
+            parameters={
+                'type': 'object',
+                'properties': {
+                    'code': {
+                        'type': 'string',
+                        'description': '要通过 python -c 执行的 Python 代码。code 和 script_path 必须二选一。',
+                    },
+                    'script_path': {
+                        'type': 'string',
+                        'description': '工作区内已有 Python 脚本路径。code 和 script_path 必须二选一。',
+                    },
+                    'args': {
+                        'type': 'array',
+                        'items': {'type': 'string'},
+                        'description': '传给脚本的命令行参数。仅在 script_path 模式下使用。',
+                    },
+                    'stdin': {
+                        'type': 'string',
+                        'description': '可选标准输入内容。',
+                    },
+                    'timeout_seconds': {
+                        'type': 'number',
+                        'minimum': 1,
+                        'description': '可选超时时间，默认使用当前会话命令超时。',
+                    },
+                    'max_output_chars': {
+                        'type': 'integer',
+                        'minimum': 100,
+                        'description': '可选输出截断长度，默认使用当前会话输出限制。',
+                    },
+                },
+            },
+            handler=_run_python_exec,
+        ),
         AgentTool(
             name='bash',
-            description='Run a shell command in the workspace. Use sparingly and prefer dedicated file tools for edits.',
+            description=(
+                '运行真实 shell 命令，例如系统命令、进程控制、git 只读检查或用户已确认的依赖安装。'
+                '不要用 bash 执行 python、python3 或 .venv/bin/python；结构化文件分析、'
+                'JSON/JSONL 处理、批量校验、数据抽样和快速计算必须优先使用 python_exec。'
+            ),
             parameters={
                 'type': 'object',
                 'properties': {
@@ -2221,11 +2269,15 @@ def _ensure_write_allowed(context: ToolExecutionContext) -> None:
         )
 
 
-def _ensure_shell_allowed(command: str, context: ToolExecutionContext) -> None:
+def _ensure_process_execution_allowed(context: ToolExecutionContext, tool_label: str) -> None:
     if not context.permissions.allow_shell_commands:
         raise ToolPermissionError(
-            'Shell commands are disabled. Re-run with --allow-shell to enable bash.'
+            f'{tool_label} is disabled. Re-run with --allow-shell to enable process execution.'
         )
+
+
+def _ensure_shell_allowed(command: str, context: ToolExecutionContext) -> None:
+    _ensure_process_execution_allowed(context, 'Shell commands')
     if context.permissions.allow_destructive_shell_commands:
         return
     destructive_patterns = [
@@ -2503,6 +2555,112 @@ def _grep_search(arguments: dict[str, Any], context: ToolExecutionContext) -> st
     return '\n'.join(hits) if hits else '(no matches)'
 
 
+def _run_python_exec(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
+    _ensure_process_execution_allowed(context, 'Python execution')
+    code = _optional_string(arguments, 'code')
+    script_path = _optional_string(arguments, 'script_path')
+    if bool(code) == bool(script_path):
+        raise ToolExecutionError('code and script_path must specify exactly one value')
+
+    raw_args = arguments.get('args', [])
+    if raw_args is None:
+        raw_args = []
+    if not isinstance(raw_args, list) or not all(isinstance(item, str) for item in raw_args):
+        raise ToolExecutionError('args must be an array of strings')
+
+    stdin = _optional_string(arguments, 'stdin')
+    timeout_seconds = _coerce_float(
+        arguments,
+        'timeout_seconds',
+        context.command_timeout_seconds,
+    )
+    max_output_chars = _coerce_int(arguments, 'max_output_chars', context.max_output_chars)
+    interpreter = _resolve_python_interpreter(context)
+    if code:
+        command = [interpreter, '-c', code]
+        mode = 'code'
+    else:
+        script = _resolve_path(script_path, context)
+        if not script.exists():
+            raise ToolExecutionError(f'Python script not found: {script_path}')
+        if not script.is_file():
+            raise ToolExecutionError(f'Python script path is not a file: {script_path}')
+        command = [interpreter, str(script), *raw_args]
+        mode = 'script'
+
+    try:
+        completed = subprocess.run(
+            command,
+            cwd=context.root,
+            input=stdin if stdin else None,
+            capture_output=True,
+            text=True,
+            timeout=timeout_seconds,
+            env=_build_subprocess_env(context),
+        )
+    except subprocess.TimeoutExpired as exc:
+        stdout = exc.stdout if isinstance(exc.stdout, str) else ''
+        stderr = exc.stderr if isinstance(exc.stderr, str) else ''
+        payload = [
+            'timed_out=true',
+            f'timeout_seconds={timeout_seconds:g}',
+            '[stdout]',
+            stdout.rstrip(),
+            '[stderr]',
+            stderr.rstrip(),
+        ]
+        return (
+            _truncate_output('\n'.join(payload).strip(), max_output_chars),
+            {
+                'action': 'python_exec',
+                'mode': mode,
+                'interpreter': interpreter,
+                'timed_out': True,
+                'timeout_seconds': timeout_seconds,
+                'stdout_preview': _snapshot_text(stdout),
+                'stderr_preview': _snapshot_text(stderr),
+                'output_preview': _snapshot_text('\n'.join(payload).strip()),
+            },
+        )
+
+    stdout = completed.stdout or ''
+    stderr = completed.stderr or ''
+    payload = [
+        f'exit_code={completed.returncode}',
+        f'interpreter={interpreter}',
+        '[stdout]',
+        stdout.rstrip(),
+        '[stderr]',
+        stderr.rstrip(),
+    ]
+    return (
+        _truncate_output('\n'.join(payload).strip(), max_output_chars),
+        {
+            'action': 'python_exec',
+            'mode': mode,
+            'interpreter': interpreter,
+            'exit_code': completed.returncode,
+            'stdout_preview': _snapshot_text(stdout),
+            'stderr_preview': _snapshot_text(stderr),
+            'output_preview': _snapshot_text('\n'.join(payload).strip()),
+        },
+    )
+
+
+def _resolve_python_interpreter(context: ToolExecutionContext) -> str:
+    # 优先使用项目虚拟环境，保证数据处理依赖和 agent 运行环境一致。
+    venv_python = context.root / '.venv' / 'bin' / 'python'
+    if venv_python.exists() and os.access(venv_python, os.X_OK):
+        return str(venv_python)
+    python3 = shutil.which('python3')
+    if python3:
+        return python3
+    python = shutil.which('python')
+    if python:
+        return python
+    return sys.executable
+
+
 def _run_bash(arguments: dict[str, Any], context: ToolExecutionContext) -> str:
     command = _require_string(arguments, 'command')
     _ensure_shell_allowed(command, context)