execution-tools¶
第4章 · 工具 · 配套项目
chapter4/execution-tools
项目说明¶
Execution Tools MCP Server¶
An MCP (Model Context Protocol) server that provides comprehensive execution tools with built-in safety mechanisms for AI agents.
Features¶
Safety Mechanisms¶
- LLM-Based Approval: Irreversible operations require approval from a secondary LLM before execution
- Result Summarization: Execution tool outputs larger than 10,000 characters are automatically summarized by an LLM for easier processing
- Automatic Verification: Operations that can be verified (e.g., syntax checking) are automatically validated
Tool Categories¶
File System Tools¶
- file_write: Write content to files with automatic syntax verification
- file_edit: Edit existing files with diff preview and verification
Generic Execution Tools¶
- code_interpreter: Execute Python code in a sandboxed environment with result analysis
- virtual_terminal: Execute shell commands with error summarization
External System Integration Tools¶
- google_calendar_add: Add events to Google Calendar
- github_create_pr: Create GitHub Pull Requests with validation
Installation¶
Configuration¶
-
Copy
env.exampleto.env: -
Configure your environment variables:
# LLM Configuration (for safety checks and summarization) PROVIDER=kimi # API Keys (set the one for your provider) KIMI_API_KEY=your_kimi_key # SILICONFLOW_API_KEY=your_siliconflow_key # DOUBAO_API_KEY=your_doubao_key # OPENROUTER_API_KEY=your_openrouter_key # Model (optional, defaults to provider's default) # MODEL=kimi-k3 # Model parameters TEMPERATURE=0.7 MAX_TOKENS=4096 # External Services (optional) GOOGLE_CALENDAR_CREDENTIALS_FILE=credentials.json GITHUB_TOKEN=your_github_token # Safety Settings REQUIRE_APPROVAL_FOR_DANGEROUS_OPS=true AUTO_SUMMARIZE_COMPLEX_OUTPUT=true AUTO_VERIFY_CODE=true
Supported Providers:
- siliconflow: Qwen/Qwen3-235B-A22B-Thinking-2507
- doubao: doubao-seed-1-6-thinking-250715
- kimi/moonshot: kimi-k3
- openrouter: google/gemini-3.5-flash (or openai/gpt-5.6-luna, anthropic/claude-sonnet-4.6)
Universal OpenRouter fallback: when the configured
PROVIDER's key is missing butOPENROUTER_API_KEYis set, the LLM steps (approval, summarization, error/syntax analysis) transparently switch toopenrouterviaConfig.effective_provider(). SetMODELto aprovider/modelid for OpenRouter, e.g.MODEL=openai/gpt-5.6-luna.
Usage¶
命令行入口 (CLI)¶
cli.py 是统一的命令行入口，用于列出、单独调用每个执行工具，并运行端到端演示。
它复用与 MCP 服务器相同的工具实现，因此行为完全一致。
# 查看总帮助与所有子命令
python cli.py --help
# 列出所有执行工具
python cli.py list
# 端到端离线演示（推荐先看这个；无需 API key 即可运行）
python cli.py demo
# 单独调用某个工具
python cli.py code --language python --code "print(2 ** 10)"
python cli.py shell "python3 --version"
python cli.py write --path notes.txt --content "hello" --overwrite
python cli.py edit --path notes.txt --search hello --replace world
全局开关（放在子命令之前）：
| 开关 | 作用 |
|---|---|
--provider |
覆盖 LLM 提供商（PROVIDER） |
--workspace |
覆盖工作目录（文件操作被限制在此目录内） |
--no-approval |
关闭危险操作的 LLM 事前审批 |
--no-verify |
关闭写文件/代码的自动语法校验 |
--no-summarize |
关闭长输出的 LLM 总结（仍会截断并持久化） |
离线运行：list、demo 以及关闭了审批/总结/非 Python 校验的
code/shell/write/edit 均无需 API key。需要 API key 的场景为：LLM 事前审批、
长输出的 LLM 总结、非 Python 语法校验。calendar 与 pr 还额外需要相应外部凭据。
长输出的截断与持久化：当
code_interpreter/virtual_terminal的输出 超过阈值（默认 200 行或 10000 字符）时，工具只在上下文中保留头尾各 50 行， 完整输出落盘到临时文件，并在返回值的stdout_file/stderr_file字段给出路径。 该机制不依赖 LLM，可离线工作。
Running the MCP Server¶
Using with MCP Client¶
from mcp import Client
# Connect to the MCP server
client = Client("stdio://python server.py")
# Use file write tool
result = await client.call_tool("file_write", {
"path": "test.py",
"content": "print('Hello, World!')"
})
# Use code interpreter
result = await client.call_tool("code_interpreter", {
"code": "import math\nprint(math.sqrt(16))"
})
# Use virtual terminal
result = await client.call_tool("virtual_terminal", {
"command": "ls -la"
})
Testing Individual Tools¶
# Test file operations
python test_file_tools.py
# Test execution tools
python test_execution_tools.py
# Test external integrations
python test_external_tools.py
Architecture¶
The server implements a layered architecture:
- Safety Layer: Intercepts dangerous operations and validates them
- Tool Layer: Implements individual tool logic
- Verification Layer: Validates outputs and provides feedback
- Integration Layer: Connects to external services
Examples¶
See examples.py for comprehensive usage examples.
实验 4-2：执行工具 MCP 服务器¶
本项目对应书中第 4 章「执行工具」一节的实验 4-2，聚焦执行工具的安全机制：
分层安全防护（输入验证、权限控制、LLM 事前审批）、自动语法验证与反馈闭环、
以及长输出的截断与持久化。推荐从 python cli.py demo 开始。
源代码¶
cli.py¶
#!/usr/bin/env python3
"""执行工具统一命令行入口（实验 4-2：执行工具 MCP 服务器）。
本文件提供一个 argparse 命令行界面，用于列出、单独调用每个执行工具，并运行
一个端到端的离线演示。它复用 server.py 背后的同一批工具实现，因此命令行的
行为与 MCP 服务器完全一致。
工具清单（与 server.py 一致）：
file_write 写文件（写入前自动做语法/linter 校验）
file_edit 按“搜索-替换”编辑文件（带 diff 预览与校验）
code_interpreter 多语言沙盒代码执行（危险操作审批、长输出截断持久化）
virtual_terminal Shell 命令执行（危险命令检测、长输出截断持久化）
google_calendar_add 创建 Google 日历事件（需要凭据）
github_create_pr 创建 GitHub Pull Request（需要 token）
安全机制（与书中“执行工具”一节对应）：
- LLM 事前审批：不可逆/危险操作在执行前交由独立 LLM 审查
- 自动验证：Python 语法通过 compile() 本地校验，其他语言由 LLM 兜底
- 长输出截断与持久化：超过阈值时仅保留头尾若干行，完整输出落盘到临时文件
用法示例：
python cli.py list
python cli.py demo
python cli.py code --language python --code "print(2 ** 10)"
python cli.py shell "python3 --version"
python cli.py write --path notes.txt --content "hello" --overwrite
python cli.py --no-approval --no-summarize shell "ls -la"
不需要 API key 的命令：list、demo（离线路径）、以及关闭了审批/总结/非 Python
校验的 code/shell/write/edit。需要 API key 的场景：LLM 审批、长输出 LLM 总结、
非 Python 语法校验。calendar 与 pr 还额外需要相应的外部凭据。
"""
import argparse
import asyncio
import json
import os
import sys
import tempfile
import textwrap
# ---------------------------------------------------------------------------
# 工具元数据（供 `list` 子命令展示）
# ---------------------------------------------------------------------------
TOOL_CATALOG = [
("file_write", "文件系统", "写文件，写入前自动做语法/linter 校验"),
("file_edit", "文件系统", "按搜索-替换编辑文件，附带 diff 预览与校验"),
("code_interpreter", "通用执行", "多语言沙盒代码执行（Python/JS/Go/Java/C++/Rust/PHP/Bash）"),
("virtual_terminal", "通用执行", "Shell 命令执行，含危险命令检测与长输出截断"),
("google_calendar_add", "外部系统", "创建 Google 日历事件（需要 credentials.json）"),
("github_create_pr", "外部系统", "创建 GitHub Pull Request（需要 GITHUB_TOKEN）"),
]
def _apply_global_env(args: argparse.Namespace) -> None:
"""把全局开关写入环境变量，供 config.py 在导入时读取。
config.Config 在模块导入时读取环境变量，因此所有涉及配置的模块都必须在此
函数执行之后才导入（本文件中的工具模块均为函数内延迟导入）。
"""
if args.provider:
os.environ["PROVIDER"] = args.provider
if args.workspace:
os.environ["WORKSPACE_DIR"] = os.path.abspath(args.workspace)
if args.no_approval:
os.environ["REQUIRE_APPROVAL_FOR_DANGEROUS_OPS"] = "false"
if args.no_verify:
os.environ["AUTO_VERIFY_CODE"] = "false"
if args.no_summarize:
os.environ["AUTO_SUMMARIZE_COMPLEX_OUTPUT"] = "false"
def _build_tools():
"""构造共享的工具实例（延迟导入，确保环境变量已就绪）。"""
from llm_helper import LLMHelper
from file_tools import FileTools
from execution_tools import ExecutionTools
from external_tools import ExternalTools
llm_helper = LLMHelper() # 客户端惰性创建，离线时不需要 API key
return {
"llm": llm_helper,
"file": FileTools(llm_helper),
"exec": ExecutionTools(llm_helper),
"external": ExternalTools(llm_helper),
}
def _print_result(result: dict) -> None:
"""统一以 JSON 打印工具返回结果。"""
print(json.dumps(result, indent=2, ensure_ascii=False))
# ---------------------------------------------------------------------------
# 子命令实现
# ---------------------------------------------------------------------------
def cmd_list(args: argparse.Namespace) -> int:
print("可用执行工具：\n")
print(f" {'工具名':<20} {'类别':<8} 说明")
print(f" {'-' * 20} {'-' * 8} {'-' * 40}")
for name, category, desc in TOOL_CATALOG:
print(f" {name:<20} {category:<8} {desc}")
print("\n用 `python cli.py <子命令> --help` 查看每个工具的参数。")
print("用 `python cli.py demo` 运行端到端离线演示。")
return 0
def cmd_code(args: argparse.Namespace) -> int:
code = args.code
if args.file:
with open(args.file, "r", encoding="utf-8") as f:
code = f.read()
if not code:
print("错误：请通过 --code 或 --file 提供要执行的代码。", file=sys.stderr)
return 2
tools = _build_tools()
result = asyncio.run(tools["exec"].code_interpreter(
code=code,
language=args.language,
timeout=args.timeout,
stdin=args.stdin,
))
_print_result(result)
return 0 if result.get("success") else 1
def cmd_shell(args: argparse.Namespace) -> int:
tools = _build_tools()
result = asyncio.run(tools["exec"].virtual_terminal(
command=args.command,
timeout=args.timeout,
))
_print_result(result)
return 0 if result.get("success") else 1
def cmd_write(args: argparse.Namespace) -> int:
content = args.content
if args.content_file:
with open(args.content_file, "r", encoding="utf-8") as f:
content = f.read()
if content is None:
print("错误：请通过 --content 或 --content-file 提供文件内容。", file=sys.stderr)
return 2
tools = _build_tools()
result = asyncio.run(tools["file"].write_file(
path=args.path,
content=content,
overwrite=args.overwrite,
))
_print_result(result)
return 0 if result.get("success") else 1
def cmd_edit(args: argparse.Namespace) -> int:
tools = _build_tools()
result = asyncio.run(tools["file"].edit_file(
path=args.path,
search=args.search,
replace=args.replace,
))
_print_result(result)
return 0 if result.get("success") else 1
def cmd_calendar(args: argparse.Namespace) -> int:
tools = _build_tools()
result = asyncio.run(tools["external"].google_calendar_add(
summary=args.summary,
start_time=args.start,
end_time=args.end,
description=args.description,
location=args.location,
))
_print_result(result)
return 0 if result.get("success") else 1
def cmd_pr(args: argparse.Namespace) -> int:
tools = _build_tools()
result = asyncio.run(tools["external"].github_create_pr(
repo_name=args.repo,
title=args.title,
body=args.body,
head_branch=args.head,
base_branch=args.base,
))
_print_result(result)
return 0 if result.get("success") else 1
def cmd_demo(args: argparse.Namespace) -> int:
"""端到端离线演示：模拟一个 Agent 用执行工具完成一个真实小任务。
场景：Agent 需要写一个词频统计脚本、生成样本数据、运行统计、再用 shell
校验结果。演示同时覆盖四个安全机制：linter 校验、危险命令 fail-safe 审批、
长输出截断与持久化。整个流程默认离线运行（关闭 LLM 总结）。
"""
# 演示放在独立临时工作区，避免污染当前目录。
workspace = tempfile.mkdtemp(prefix="exec_tools_demo_")
os.environ["WORKSPACE_DIR"] = workspace
# 离线运行：关闭需要 LLM 的输出总结（截断持久化不依赖 LLM）。
if "AUTO_SUMMARIZE_COMPLEX_OUTPUT" not in os.environ:
os.environ["AUTO_SUMMARIZE_COMPLEX_OUTPUT"] = "false"
tools = _build_tools()
file_tools = tools["file"]
exec_tools = tools["exec"]
def section(title: str) -> None:
print("\n" + "=" * 64)
print(title)
print("=" * 64)
print(f"演示工作区：{workspace}")
print("（离线路径，无需 API key；如已配置 key，审批/总结将走真实 LLM）")
async def run() -> None:
# 1. 写文件 + 自动 linter 校验（合法代码）
section("1. file_write：写入词频统计脚本（自动语法校验）")
script = textwrap.dedent('''\
"""统计文本文件中的词频。"""
import sys
from collections import Counter
def word_count(path):
with open(path, encoding="utf-8") as f:
words = f.read().split()
return Counter(words)
if __name__ == "__main__":
for word, freq in word_count(sys.argv[1]).most_common(5):
print(f"{word}\\t{freq}")
''')
r = await file_tools.write_file("wordcount.py", script, overwrite=True)
print(f"结果：success={r['success']}, verification={r.get('verification')}")
print(f"写入：{r.get('path')}")
# 2. linter 拦截语法错误的代码
section("2. file_write：写入含语法错误的代码（linter 应拦截）")
broken = "def broken(:\n return 1\n"
r = await file_tools.write_file("broken.py", broken, overwrite=True)
print(f"结果：success={r['success']}")
print(f"校验反馈：{r.get('error')}")
# 3. 生成样本数据
section("3. file_write：生成样本数据文件")
sample = "apple banana apple cherry banana apple date cherry banana apple\n"
r = await file_tools.write_file("data.txt", sample, overwrite=True)
print(f"结果：success={r['success']}，写入 {r.get('bytes_written')} 字节")
# 4. code_interpreter：运行统计脚本
section("4. code_interpreter：运行统计逻辑（Python 沙盒）")
analysis = textwrap.dedent('''\
from collections import Counter
text = "apple banana apple cherry banana apple date cherry banana apple"
for word, freq in Counter(text.split()).most_common(3):
print(f"{word}: {freq}")
''')
r = await exec_tools.code_interpreter(code=analysis, language="python")
print(f"结果：success={r['success']}, returncode={r.get('returncode')}")
print("stdout:")
print(textwrap.indent(r.get("stdout", ""), " "))
# 5. virtual_terminal：用 shell 校验数据文件
section("5. virtual_terminal：用 shell 校验数据文件")
r = await exec_tools.virtual_terminal(
command=f"wc -w {workspace}/data.txt && echo '--- 词数统计完成 ---'"
)
print(f"结果：success={r['success']}, returncode={r.get('returncode')}")
print("stdout:")
print(textwrap.indent(r.get("stdout", ""), " "))
# 6. 长输出截断与持久化（离线，不需 LLM）
section("6. code_interpreter：长输出自动截断并落盘")
long_code = "for i in range(1000):\n print(f'line {i}: ' + 'x' * 20)\n"
r = await exec_tools.code_interpreter(code=long_code, language="python")
stdout = r.get("stdout", "")
print(f"上下文中保留的输出行数：{len(stdout.splitlines())}（原始 1000 行）")
print(f"完整输出落盘文件：{r.get('stdout_file')}")
print("上下文中输出的尾部片段：")
print(textwrap.indent("\n".join(stdout.splitlines()[-4:]), " "))
# 7. 危险命令的审批（离线 fail-safe / 在线交由真实 LLM 判断）
section("7. virtual_terminal：危险命令触发审批")
os.environ["REQUIRE_APPROVAL_FOR_DANGEROUS_OPS"] = "true"
# 目标是不存在的临时路径，即便被执行也无副作用。
danger = await exec_tools.virtual_terminal(
command="rm -rf /tmp/exec_tools_demo_nonexistent_path_xyz"
)
print(f"结果：success={danger['success']}")
if danger.get("error"):
print(f"说明：{danger.get('error')}")
print("（审批未通过：危险命令被拦截、未执行。离线无 LLM 时按 fail-safe 拒绝，"
"在线时也可能被真实 LLM 判定为高风险而拒绝。）")
else:
print("（审批通过：已配置 API key，真实 LLM 判定该命令针对不存在路径、无副作用而放行。）")
section("演示完成")
print("覆盖的安全机制：自动 linter 校验、危险命令审批、长输出截断持久化。")
print(f"演示产物位于：{workspace}")
asyncio.run(run())
return 0
# ---------------------------------------------------------------------------
# 参数解析
# ---------------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="cli.py",
description="执行工具统一命令行入口（实验 4-2：执行工具 MCP 服务器）。",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=textwrap.dedent("""\
示例：
python cli.py list 列出所有执行工具
python cli.py demo 运行端到端离线演示
python cli.py code --code "print(6*7)" 执行 Python 代码
python cli.py shell "ls -la" 执行 shell 命令
python cli.py write --path a.txt --content hi --overwrite
python cli.py --no-approval shell "echo hello"
关闭 --no-approval / --no-summarize / --no-verify 后，
code/shell/write/edit 等命令可完全离线运行，无需 API key。
"""),
)
# 全局开关
parser.add_argument("--provider", help="LLM 提供商（覆盖 PROVIDER，如 kimi/doubao/siliconflow/openrouter）")
parser.add_argument("--workspace", help="工作目录（覆盖 WORKSPACE_DIR，文件操作被限制在此目录内）")
parser.add_argument("--no-approval", action="store_true", help="关闭危险操作的 LLM 事前审批")
parser.add_argument("--no-verify", action="store_true", help="关闭写文件/代码的自动语法校验")
parser.add_argument("--no-summarize", action="store_true", help="关闭长输出的 LLM 总结（仍会截断持久化）")
sub = parser.add_subparsers(dest="command", metavar="<子命令>")
p = sub.add_parser("list", help="列出所有可用的执行工具")
p.set_defaults(func=cmd_list)
p = sub.add_parser("demo", help="运行端到端离线演示（推荐先看这个）")
p.set_defaults(func=cmd_demo)
p = sub.add_parser("code", help="调用 code_interpreter 执行代码")
p.add_argument("--code", help="要执行的代码字符串")
p.add_argument("--file", help="从文件读取要执行的代码")
p.add_argument("--language", default="python",
help="编程语言（python/javascript/typescript/go/java/cpp/rust/php/bash，默认 python）")
p.add_argument("--timeout", type=float, default=30.0, help="执行超时秒数（默认 30）")
p.add_argument("--stdin", help="可选的标准输入")
p.set_defaults(func=cmd_code)
p = sub.add_parser("shell", help="调用 virtual_terminal 执行 shell 命令")
p.add_argument("command", help="要执行的 shell 命令")
p.add_argument("--timeout", type=int, default=30, help="超时秒数（默认 30）")
p.set_defaults(func=cmd_shell)
p = sub.add_parser("write", help="调用 file_write 写文件")
p.add_argument("--path", required=True, help="文件路径（相对工作目录或绝对路径）")
p.add_argument("--content", help="文件内容")
p.add_argument("--content-file", help="从文件读取要写入的内容")
p.add_argument("--overwrite", action="store_true", help="允许覆盖已存在文件")
p.set_defaults(func=cmd_write)
p = sub.add_parser("edit", help="调用 file_edit 按搜索-替换编辑文件")
p.add_argument("--path", required=True, help="文件路径")
p.add_argument("--search", required=True, help="要搜索的文本")
p.add_argument("--replace", required=True, help="替换文本")
p.set_defaults(func=cmd_edit)
p = sub.add_parser("calendar", help="调用 google_calendar_add 创建日历事件（需要凭据）")
p.add_argument("--summary", required=True, help="事件标题")
p.add_argument("--start", required=True, help="开始时间（ISO 8601，如 2025-10-01T10:00:00）")
p.add_argument("--end", required=True, help="结束时间（ISO 8601）")
p.add_argument("--description", help="事件描述")
p.add_argument("--location", help="事件地点")
p.set_defaults(func=cmd_calendar)
p = sub.add_parser("pr", help="调用 github_create_pr 创建 Pull Request（需要 token）")
p.add_argument("--repo", required=True, help="仓库名（owner/repo 格式）")
p.add_argument("--title", required=True, help="PR 标题")
p.add_argument("--body", required=True, help="PR 描述")
p.add_argument("--head", required=True, help="源分支")
p.add_argument("--base", default="main", help="目标分支（默认 main）")
p.set_defaults(func=cmd_pr)
return parser
def main(argv=None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
if not getattr(args, "command", None):
parser.print_help()
return 0
_apply_global_env(args)
try:
return args.func(args)
except KeyboardInterrupt:
print("\n已中断。", file=sys.stderr)
return 130
if __name__ == "__main__":
sys.exit(main())
config.py¶
"""Configuration management for the execution tools MCP server."""
import os
import sys
from pathlib import Path
from typing import Optional
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
def _env_int(name: str, default: int) -> int:
"""Read an integer env var; fall back to default (with a warning) if malformed."""
raw = os.getenv(name)
if raw is None:
return default
try:
return int(raw)
except ValueError:
print(f"Warning: invalid {name}={raw!r} (must be an integer); using default {default}",
file=sys.stderr)
return default
def _env_float(name: str, default: float) -> float:
"""Read a float env var; fall back to default (with a warning) if malformed."""
raw = os.getenv(name)
if raw is None:
return default
try:
return float(raw)
except ValueError:
print(f"Warning: invalid {name}={raw!r} (must be a number); using default {default}",
file=sys.stderr)
return default
class Config:
"""Configuration for the MCP server."""
# LLM Configuration
PROVIDER: str = os.getenv("PROVIDER", "kimi")
# API Keys
SILICONFLOW_API_KEY: Optional[str] = os.getenv("SILICONFLOW_API_KEY")
DOUBAO_API_KEY: Optional[str] = os.getenv("DOUBAO_API_KEY")
KIMI_API_KEY: Optional[str] = os.getenv("KIMI_API_KEY")
MOONSHOT_API_KEY: Optional[str] = os.getenv("MOONSHOT_API_KEY")
OPENROUTER_API_KEY: Optional[str] = os.getenv("OPENROUTER_API_KEY")
# Model names (optional, defaults to provider defaults)
MODEL: Optional[str] = os.getenv("MODEL")
# Model parameters
TEMPERATURE: float = _env_float("TEMPERATURE", 0.7)
MAX_TOKENS: int = _env_int("MAX_TOKENS", 4096)
# External Services
GOOGLE_CALENDAR_CREDENTIALS_FILE: str = os.getenv(
"GOOGLE_CALENDAR_CREDENTIALS_FILE",
"credentials.json"
)
GITHUB_TOKEN: Optional[str] = os.getenv("GITHUB_TOKEN")
# Safety Settings
REQUIRE_APPROVAL_FOR_DANGEROUS_OPS: bool = (
os.getenv("REQUIRE_APPROVAL_FOR_DANGEROUS_OPS", "true").lower() == "true"
)
AUTO_SUMMARIZE_COMPLEX_OUTPUT: bool = (
os.getenv("AUTO_SUMMARIZE_COMPLEX_OUTPUT", "true").lower() == "true"
)
AUTO_VERIFY_CODE: bool = (
os.getenv("AUTO_VERIFY_CODE", "true").lower() == "true"
)
MAX_OUTPUT_LENGTH: int = _env_int("MAX_OUTPUT_LENGTH", 1000)
# Workspace Configuration
WORKSPACE_DIR: Path = Path(os.getenv("WORKSPACE_DIR", os.getcwd()))
@classmethod
def get_api_key(cls, provider: str) -> Optional[str]:
"""Get API key for the specified provider."""
provider = provider.lower()
if provider == "siliconflow":
return cls.SILICONFLOW_API_KEY
elif provider == "doubao":
return cls.DOUBAO_API_KEY
elif provider in ["kimi", "moonshot"]:
return cls.KIMI_API_KEY or cls.MOONSHOT_API_KEY
elif provider == "openrouter":
return cls.OPENROUTER_API_KEY
return None
@classmethod
def effective_provider(cls) -> str:
"""Resolve the provider actually used, applying the OpenRouter fallback.
Preserves default behavior when the configured provider's key is
present. Otherwise, if an OPENROUTER_API_KEY is available, transparently
fall back to 'openrouter' so the tools still run with only that key set.
"""
provider = cls.PROVIDER.lower()
if cls.get_api_key(provider):
return provider
if cls.OPENROUTER_API_KEY:
return "openrouter"
return provider
@classmethod
def validate(cls) -> None:
"""Validate the configuration."""
provider = cls.effective_provider()
api_key = cls.get_api_key(provider)
if not api_key:
raise ValueError(
f"API key required for provider '{cls.PROVIDER.lower()}'. "
f"Set one of {cls.PROVIDER.upper()}_API_KEY or OPENROUTER_API_KEY "
f"(universal fallback)."
)
@classmethod
def get_llm_config(cls) -> dict:
"""Get LLM configuration based on provider."""
provider = cls.effective_provider()
api_key = cls.get_api_key(provider)
if not api_key:
raise ValueError(
f"API key not found for provider '{cls.PROVIDER.lower()}'. "
f"Set {cls.PROVIDER.upper()}_API_KEY or OPENROUTER_API_KEY."
)
if provider == "siliconflow":
return {
"provider": "siliconflow",
"api_key": api_key,
"base_url": "https://api.siliconflow.cn/v1",
"model": cls.MODEL or "Qwen/Qwen3-235B-A22B-Thinking-2507"
}
elif provider == "doubao":
return {
"provider": "doubao",
"api_key": api_key,
"base_url": "https://ark.cn-beijing.volces.com/api/v3",
"model": cls.MODEL or "doubao-seed-1-6-thinking-250715"
}
elif provider in ["kimi", "moonshot"]:
return {
"provider": "kimi",
"api_key": api_key,
"base_url": "https://api.moonshot.cn/v1",
"model": cls.MODEL or "kimi-k3"
}
elif provider == "openrouter":
return {
"provider": "openrouter",
"api_key": api_key,
"base_url": "https://openrouter.ai/api/v1",
"model": cls.MODEL or "google/gemini-3.5-flash"
}
else:
raise ValueError(
f"Unsupported provider: {provider}. "
f"Use 'siliconflow', 'doubao', 'kimi', 'moonshot', or 'openrouter'"
)
# Note: configuration is validated lazily when the LLM is actually used
# (see LLMHelper), so that execution tools which do not require an LLM
# (file write, code run, terminal) can be used offline without an API key.
examples.py¶
"""Example usage of the execution tools MCP server."""
import asyncio
from llm_helper import LLMHelper
from file_tools import FileTools
from execution_tools import ExecutionTools
from external_tools import ExternalTools
async def example_file_operations():
"""Example: File operations with verification."""
print("=== File Operations Example ===\n")
llm_helper = LLMHelper()
file_tools = FileTools(llm_helper)
# Write a Python file
print("1. Writing a Python file...")
result = await file_tools.write_file(
path="test_script.py",
content="""def greet(name):
print(f"Hello, {name}!")
if __name__ == "__main__":
greet("World")
""",
overwrite=True
)
print(f"Result: {result}\n")
# Edit the file
print("2. Editing the file...")
result = await file_tools.edit_file(
path="test_script.py",
search='greet("World")',
replace='greet("MCP Server")'
)
print(f"Result: {result}\n")
async def example_code_interpreter():
"""Example: Code interpreter with analysis."""
print("=== Code Interpreter Example ===\n")
llm_helper = LLMHelper()
execution_tools = ExecutionTools(llm_helper)
# Execute valid code
print("1. Executing valid code...")
result = await execution_tools.code_interpreter(
code="""
import math
# Calculate factorial
def factorial(n):
if n <= 1:
return 1
return n * factorial(n - 1)
print(f"Factorial of 5: {factorial(5)}")
print(f"Square root of 16: {math.sqrt(16)}")
"""
)
print(f"Result: {result}\n")
# Execute code with error
print("2. Executing code with error...")
result = await execution_tools.code_interpreter(
code="""
# This will cause an error
x = 10 / 0
"""
)
print(f"Result: {result}\n")
async def example_virtual_terminal():
"""Example: Virtual terminal."""
print("=== Virtual Terminal Example ===\n")
llm_helper = LLMHelper()
execution_tools = ExecutionTools(llm_helper)
# Execute simple command
print("1. Listing current directory...")
result = await execution_tools.virtual_terminal(
command="ls -la"
)
print(f"Result: {result}\n")
# Execute command with error
print("2. Executing command that fails...")
result = await execution_tools.virtual_terminal(
command="cat nonexistent_file.txt"
)
print(f"Result: {result}\n")
async def example_google_calendar():
"""Example: Google Calendar integration."""
print("=== Google Calendar Example ===\n")
llm_helper = LLMHelper()
external_tools = ExternalTools(llm_helper)
print("Adding event to Google Calendar...")
result = await external_tools.google_calendar_add(
summary="Team Meeting",
start_time="2025-10-01T10:00:00",
end_time="2025-10-01T11:00:00",
description="Quarterly planning meeting",
location="Conference Room A"
)
print(f"Result: {result}\n")
async def example_github_pr():
"""Example: GitHub Pull Request creation."""
print("=== GitHub PR Example ===\n")
llm_helper = LLMHelper()
external_tools = ExternalTools(llm_helper)
print("Creating GitHub Pull Request...")
result = await external_tools.github_create_pr(
repo_name="owner/repository",
title="Add new feature",
body="This PR adds a new feature to improve performance.\n\n## Changes\n- Optimized algorithm\n- Added tests\n- Updated documentation",
head_branch="feature/new-feature",
base_branch="main"
)
print(f"Result: {result}\n")
async def main():
"""Run all examples."""
try:
# File operations
await example_file_operations()
# Code interpreter
await example_code_interpreter()
# Virtual terminal
await example_virtual_terminal()
# External tools (commented out as they require credentials)
# await example_google_calendar()
# await example_github_pr()
print("All examples completed!")
except Exception as e:
print(f"Error running examples: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(main())
execution_tools.py¶
"""Generic execution tools: code interpreter and virtual terminal."""
import os
import subprocess
import sys
import io
import tempfile
import traceback
from typing import Dict, Any, Optional, Tuple
from contextlib import redirect_stdout, redirect_stderr
from llm_helper import LLMHelper
from config import Config
from multilang_executor import LanguageExecutor, ExecutionStatus
# Long-output handling thresholds (see "长输出的截断与持久化" in chapter 4).
# When output exceeds either threshold, keep the head and tail few lines in the
# context and persist the full output to a temp file for later retrieval.
MAX_OUTPUT_LINES = 200
MAX_OUTPUT_CHARS = 10000
HEAD_LINES = 50
TAIL_LINES = 50
def truncate_and_persist(
text: str,
tool_name: str = "execution",
max_lines: int = MAX_OUTPUT_LINES,
max_chars: int = MAX_OUTPUT_CHARS,
head_lines: int = HEAD_LINES,
tail_lines: int = TAIL_LINES,
) -> Tuple[str, Optional[str]]:
"""Truncate over-long output and persist the full text to a temp file.
Returns a tuple of (processed_text, saved_path). When the output is within
both thresholds, it is returned unchanged with ``saved_path`` set to None.
Otherwise only the first ``head_lines`` and last ``tail_lines`` lines are
kept in context, with a middle marker pointing to the saved file. This
keeps the agent's context bounded without discarding any information and
requires no LLM call.
"""
if text is None:
return text, None
lines = text.split("\n")
if len(text) <= max_chars and len(lines) <= max_lines:
return text, None
# Persist the complete output for later retrieval via read_file.
fd, path = tempfile.mkstemp(prefix=f"{tool_name}_output_", suffix=".txt")
with os.fdopen(fd, "w", encoding="utf-8") as f:
f.write(text)
head_part = lines[:head_lines]
tail_part = lines[-tail_lines:]
omitted = max(len(lines) - head_lines - tail_lines, 0)
middle = f"... [省略 {omitted} 行，完整输出已保存至 {path}] ..."
guide = f"[如需完整输出，请使用 read_file 工具读取 {path}]"
truncated = "\n".join(head_part + [middle] + tail_part + [guide])
return truncated, path
class ExecutionTools:
"""Generic execution tools with safety and result analysis."""
def __init__(self, llm_helper: LLMHelper):
"""Initialize execution tools with LLM helper."""
self.llm_helper = llm_helper
self.lang_executor = LanguageExecutor(workspace_dir=Config.WORKSPACE_DIR)
async def code_interpreter(
self,
code: str,
language: str = "python",
timeout: float = 30.0,
stdin: Optional[str] = None,
files: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
"""
Execute code in a sandboxed environment with multi-language support.
Args:
code: Code to execute
language: Programming language (python, javascript, typescript, go, java, cpp, rust, php, bash)
timeout: Execution timeout in seconds
stdin: Optional stdin input
files: Optional additional files
Returns:
Result dictionary with output and analysis
"""
if language is None:
language = "python"
language = language.lower()
# Verify syntax first (only for Python for now)
if Config.AUTO_VERIFY_CODE and language in ['python', 'python3']:
is_valid, error_msg = self.llm_helper.verify_code_syntax(code, language)
if not is_valid:
return {
"success": False,
"error": f"Syntax error: {error_msg}",
"verification": "failed",
"language": language
}
# Check for dangerous operations
if Config.REQUIRE_APPROVAL_FOR_DANGEROUS_OPS:
dangerous_patterns = {
'python': ['os.system', 'subprocess', 'eval', 'exec', 'open(', '__import__', 'compile'],
'bash': ['rm -rf', 'dd if=', 'mkfs', '> /dev/', 'curl', 'wget'],
'php': ['exec(', 'system(', 'shell_exec(', 'passthru(', 'eval('],
}
patterns = dangerous_patterns.get(language, [])
detected = [p for p in patterns if p in code]
if detected:
approved, reason = self.llm_helper.request_approval(
"code_execution",
{
"code": code,
"language": language,
"detected_patterns": detected
}
)
if not approved:
return {
"success": False,
"error": f"Execution not approved: {reason}",
"language": language
}
# Execute code using multi-language executor
try:
result = await self.lang_executor.execute_code(
code=code,
language=language,
timeout=timeout,
stdin=stdin,
files=files
)
# Convert status to success flag
success = result.get('status') == ExecutionStatus.SUCCESS
# Long outputs: truncate head/tail and persist the full text to a
# temp file (offline-safe), then optionally LLM-summarize whatever
# still exceeds the char threshold.
stdout = result.get('stdout', '')
stderr = result.get('stderr', '')
stdout, stdout_file = truncate_and_persist(stdout, "code_interpreter")
stderr, stderr_file = truncate_and_persist(stderr, "code_interpreter")
if Config.AUTO_SUMMARIZE_COMPLEX_OUTPUT and len(stdout) > MAX_OUTPUT_CHARS:
stdout = self.llm_helper.summarize_output("code_interpreter", stdout)
if Config.AUTO_SUMMARIZE_COMPLEX_OUTPUT and len(stderr) > MAX_OUTPUT_CHARS:
stderr = self.llm_helper.summarize_output("code_interpreter", stderr)
return {
"success": success,
"status": result.get('status'),
"language": result.get('language', language),
"stdout": stdout,
"stderr": stderr,
"stdout_file": stdout_file,
"stderr_file": stderr_file,
"returncode": result.get('returncode'),
"error": result.get('error'),
"compile_output": result.get('compile_output'),
"phase": result.get('phase'),
"verification": "passed" if Config.AUTO_VERIFY_CODE else "skipped"
}
except Exception as e:
error_output = f"{type(e).__name__}: {str(e)}\n{traceback.format_exc()}"
return {
"success": False,
"error": error_output,
"language": language
}
async def virtual_terminal(
self,
command: str,
timeout: int = 30
) -> Dict[str, Any]:
"""
Execute shell command in a virtual terminal.
Args:
command: Shell command to execute
timeout: Timeout in seconds
Returns:
Result dictionary with output and analysis
"""
# Check for dangerous commands
if Config.REQUIRE_APPROVAL_FOR_DANGEROUS_OPS:
dangerous_commands = [
'rm -rf', 'dd', 'mkfs', 'format',
'> /dev/', 'chmod -R', 'chown -R'
]
if any(dangerous in command for dangerous in dangerous_commands):
approved, reason = self.llm_helper.request_approval(
"terminal_command",
{
"command": command,
"detected_patterns": [p for p in dangerous_commands if p in command]
}
)
if not approved:
return {
"success": False,
"error": f"Command execution not approved: {reason}"
}
# Execute command
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=timeout,
cwd=Config.WORKSPACE_DIR
)
stdout = result.stdout
stderr = result.stderr
# Long output: truncate head/tail and persist to a temp file, then
# optionally LLM-summarize whatever still exceeds the char threshold.
stdout, stdout_file = truncate_and_persist(stdout, "virtual_terminal")
stderr, stderr_file = truncate_and_persist(stderr, "virtual_terminal")
if Config.AUTO_SUMMARIZE_COMPLEX_OUTPUT:
if len(stdout) > MAX_OUTPUT_CHARS:
stdout = self.llm_helper.summarize_output(
"virtual_terminal",
stdout
)
if len(stderr) > MAX_OUTPUT_CHARS:
stderr = self.llm_helper.summarize_output(
"virtual_terminal",
stderr
)
response = {
"success": result.returncode == 0,
"returncode": result.returncode,
"stdout": stdout,
"stderr": stderr,
"stdout_file": stdout_file,
"stderr_file": stderr_file
}
return response
except subprocess.TimeoutExpired:
return {
"success": False,
"error": f"Command timed out after {timeout} seconds"
}
except Exception as e:
return {
"success": False,
"error": f"Command execution failed: {str(e)}"
}
external_tools.py¶
"""External system integration tools: Google Calendar and GitHub."""
import os
import json
from typing import Dict, Any, Optional
from datetime import datetime, timedelta
from pathlib import Path
from llm_helper import LLMHelper
from config import Config
# Google Calendar imports (optional)
try:
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
GOOGLE_AVAILABLE = True
except ImportError:
GOOGLE_AVAILABLE = False
# GitHub imports (optional)
try:
from github import Github, GithubException
GITHUB_AVAILABLE = True
except ImportError:
GITHUB_AVAILABLE = False
class ExternalTools:
"""External system integration tools."""
def __init__(self, llm_helper: LLMHelper):
"""Initialize external tools with LLM helper."""
self.llm_helper = llm_helper
self._google_service = None
self._github_client = None
def _get_google_calendar_service(self):
"""Get or create Google Calendar service."""
if not GOOGLE_AVAILABLE:
raise ImportError("Google Calendar libraries not installed")
if self._google_service:
return self._google_service
SCOPES = ['https://www.googleapis.com/auth/calendar']
creds = None
token_path = Path('token.json')
creds_path = Path(Config.GOOGLE_CALENDAR_CREDENTIALS_FILE)
# Load token if exists
if token_path.exists():
creds = Credentials.from_authorized_user_file(str(token_path), SCOPES)
# Refresh or get new token
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
if not creds_path.exists():
raise FileNotFoundError(f"Credentials file not found: {creds_path}")
flow = InstalledAppFlow.from_client_secrets_file(str(creds_path), SCOPES)
creds = flow.run_local_server(port=0)
# Save token
token_path.write_text(creds.to_json())
self._google_service = build('calendar', 'v3', credentials=creds)
return self._google_service
def _get_github_client(self):
"""Get or create GitHub client."""
if not GITHUB_AVAILABLE:
raise ImportError("GitHub library not installed")
if self._github_client:
return self._github_client
if not Config.GITHUB_TOKEN:
raise ValueError("GitHub token not configured")
self._github_client = Github(Config.GITHUB_TOKEN)
return self._github_client
async def google_calendar_add(
self,
summary: str,
start_time: str,
end_time: str,
description: Optional[str] = None,
location: Optional[str] = None
) -> Dict[str, Any]:
"""
Add event to Google Calendar.
Args:
summary: Event title
start_time: Start time (ISO 8601 format or natural language)
end_time: End time (ISO 8601 format or natural language)
description: Event description
location: Event location
Returns:
Result dictionary with event details
"""
try:
service = self._get_google_calendar_service()
except Exception as e:
return {
"success": False,
"error": f"Failed to initialize Google Calendar: {str(e)}"
}
# Parse times
try:
start_dt = self._parse_datetime(start_time)
end_dt = self._parse_datetime(end_time)
except ValueError as e:
return {
"success": False,
"error": f"Invalid datetime format: {str(e)}"
}
# Validate times
if end_dt <= start_dt:
return {
"success": False,
"error": "End time must be after start time"
}
# Request approval
if Config.REQUIRE_APPROVAL_FOR_DANGEROUS_OPS:
approved, reason = self.llm_helper.request_approval(
"google_calendar_add",
{
"summary": summary,
"start_time": start_dt.isoformat(),
"end_time": end_dt.isoformat(),
"description": description
}
)
if not approved:
return {
"success": False,
"error": f"Calendar event creation not approved: {reason}"
}
# Create event
event = {
'summary': summary,
'start': {
'dateTime': start_dt.isoformat(),
'timeZone': 'UTC'
},
'end': {
'dateTime': end_dt.isoformat(),
'timeZone': 'UTC'
}
}
if description:
event['description'] = description
if location:
event['location'] = location
try:
created_event = service.events().insert(
calendarId='primary',
body=event
).execute()
return {
"success": True,
"event_id": created_event['id'],
"event_link": created_event.get('htmlLink'),
"summary": summary,
"start_time": start_dt.isoformat(),
"end_time": end_dt.isoformat()
}
except Exception as e:
return {
"success": False,
"error": f"Failed to create calendar event: {str(e)}"
}
async def github_create_pr(
self,
repo_name: str,
title: str,
body: str,
head_branch: str,
base_branch: str = "main"
) -> Dict[str, Any]:
"""
Create a GitHub Pull Request.
Args:
repo_name: Repository name (format: owner/repo)
title: PR title
body: PR description
head_branch: Source branch
base_branch: Target branch
Returns:
Result dictionary with PR details
"""
try:
github = self._get_github_client()
except Exception as e:
return {
"success": False,
"error": f"Failed to initialize GitHub client: {str(e)}"
}
# Validate repository name
if '/' not in repo_name:
return {
"success": False,
"error": "Repository name must be in format: owner/repo"
}
# Request approval
if Config.REQUIRE_APPROVAL_FOR_DANGEROUS_OPS:
approved, reason = self.llm_helper.request_approval(
"github_create_pr",
{
"repo": repo_name,
"title": title,
"head": head_branch,
"base": base_branch,
"body_preview": body[:200]
}
)
if not approved:
return {
"success": False,
"error": f"PR creation not approved: {reason}"
}
try:
# Get repository
repo = github.get_repo(repo_name)
# Verify branches exist
try:
repo.get_branch(head_branch)
repo.get_branch(base_branch)
except Exception as e:
return {
"success": False,
"error": f"Branch verification failed: {str(e)}"
}
# Create pull request
pr = repo.create_pull(
title=title,
body=body,
head=head_branch,
base=base_branch
)
return {
"success": True,
"pr_number": pr.number,
"pr_url": pr.html_url,
"title": title,
"state": pr.state,
"created_at": pr.created_at.isoformat()
}
except GithubException as e:
return {
"success": False,
"error": f"GitHub API error: {e.data.get('message', str(e))}"
}
except Exception as e:
return {
"success": False,
"error": f"Failed to create pull request: {str(e)}"
}
def _parse_datetime(self, time_str: str) -> datetime:
"""Parse datetime string in various formats."""
# Try ISO 8601 format first
formats = [
'%Y-%m-%dT%H:%M:%S',
'%Y-%m-%d %H:%M:%S',
'%Y-%m-%d %H:%M',
'%Y-%m-%d'
]
for fmt in formats:
try:
return datetime.strptime(time_str, fmt)
except ValueError:
continue
# If no format works, raise error
raise ValueError(f"Could not parse datetime: {time_str}")
file_tools.py¶
"""File system tools with safety mechanisms."""
import itertools
import os
from pathlib import Path
from typing import Dict, Any
from llm_helper import LLMHelper
from config import Config
class FileTools:
"""File system tools with verification and safety checks."""
def __init__(self, llm_helper: LLMHelper):
"""Initialize file tools with LLM helper."""
self.llm_helper = llm_helper
self.workspace_dir = Config.WORKSPACE_DIR
def _resolve_path(self, path: str) -> Path:
"""Resolve path relative to workspace."""
path_obj = Path(path)
if not path_obj.is_absolute():
path_obj = self.workspace_dir / path_obj
return path_obj.resolve()
def _is_safe_path(self, path: Path) -> bool:
"""Check if path is within workspace."""
try:
path.resolve().relative_to(self.workspace_dir.resolve())
return True
except ValueError:
return False
async def write_file(
self,
path: str,
content: str,
overwrite: bool = False
) -> Dict[str, Any]:
"""
Write content to a file with safety checks.
Args:
path: File path (relative to workspace or absolute)
content: Content to write
overwrite: Whether to overwrite existing files
Returns:
Result dictionary with status and details
"""
resolved_path = self._resolve_path(path)
# Safety check: ensure path is within workspace
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"Path {path} is outside workspace directory"
}
# Check if file exists and overwrite is not allowed
if resolved_path.exists() and not overwrite:
# Request approval for overwriting
if Config.REQUIRE_APPROVAL_FOR_DANGEROUS_OPS:
approved, reason = self.llm_helper.request_approval(
"file_overwrite",
{
"path": str(resolved_path),
"existing_size": resolved_path.stat().st_size,
"new_content_size": len(content)
}
)
if not approved:
return {
"success": False,
"error": f"Overwrite not approved: {reason}"
}
# Verify code syntax if it's a code file
if Config.AUTO_VERIFY_CODE and resolved_path.suffix in ['.py', '.js', '.ts']:
language = {'.py': 'python', '.js': 'javascript', '.ts': 'typescript'}[resolved_path.suffix]
is_valid, error_msg = self.llm_helper.verify_code_syntax(content, language)
if not is_valid:
return {
"success": False,
"error": f"Syntax validation failed: {error_msg}",
"verification": "failed"
}
# Write the file
try:
resolved_path.parent.mkdir(parents=True, exist_ok=True)
resolved_path.write_text(content)
return {
"success": True,
"path": str(resolved_path),
"bytes_written": len(content),
"verification": "passed" if Config.AUTO_VERIFY_CODE else "skipped"
}
except Exception as e:
return {
"success": False,
"error": f"Failed to write file: {str(e)}"
}
async def edit_file(
self,
path: str,
search: str,
replace: str
) -> Dict[str, Any]:
"""
Edit a file by searching and replacing content.
Args:
path: File path
search: Text to search for
replace: Replacement text
Returns:
Result dictionary with status and details
"""
resolved_path = self._resolve_path(path)
# Safety check
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"Path {path} is outside workspace directory"
}
# Check if file exists
if not resolved_path.exists():
return {
"success": False,
"error": f"File {path} does not exist"
}
# Read current content
try:
current_content = resolved_path.read_text()
except Exception as e:
return {
"success": False,
"error": f"Failed to read file: {str(e)}"
}
# Check if search text exists
if search not in current_content:
return {
"success": False,
"error": f"Search text not found in file"
}
# Perform replacement
new_content = current_content.replace(search, replace, 1)
# Generate diff preview
diff_preview = self._generate_diff(current_content, new_content)
# Verify new content if it's code
if Config.AUTO_VERIFY_CODE and resolved_path.suffix in ['.py', '.js', '.ts']:
language = {'.py': 'python', '.js': 'javascript', '.ts': 'typescript'}[resolved_path.suffix]
is_valid, error_msg = self.llm_helper.verify_code_syntax(new_content, language)
if not is_valid:
return {
"success": False,
"error": f"Syntax validation failed after edit: {error_msg}",
"diff_preview": diff_preview
}
# Write the modified content
try:
resolved_path.write_text(new_content)
return {
"success": True,
"path": str(resolved_path),
"diff_preview": diff_preview,
"verification": "passed" if Config.AUTO_VERIFY_CODE else "skipped"
}
except Exception as e:
return {
"success": False,
"error": f"Failed to write file: {str(e)}"
}
def _generate_diff(self, old_content: str, new_content: str) -> str:
"""Generate a simple diff preview."""
old_lines = old_content.split('\n')
new_lines = new_content.split('\n')
diff_lines = []
for i, (old, new) in enumerate(itertools.zip_longest(old_lines, new_lines, fillvalue=''), 1):
if old != new:
diff_lines.append(f"Line {i}:")
diff_lines.append(f" - {old}")
diff_lines.append(f" + {new}")
return '\n'.join(diff_lines[:20]) # Limit to 20 lines
filesystem_enhanced.py¶
"""
Enhanced filesystem tools based on AWorld filesystem-server.
Provides comprehensive file and directory operations with safety checks.
"""
import json
import os
import shutil
import traceback
from pathlib import Path
from typing import Dict, Any, List
from config import Config
class FilesystemEnhanced:
"""Enhanced filesystem operations with safety and validation."""
def __init__(self):
self.workspace_dir = Path(Config.WORKSPACE_DIR).resolve()
self.allowed_directories = [self.workspace_dir]
def _resolve_path(self, path: str) -> Path:
"""Resolve path relative to workspace."""
path_obj = Path(path)
if not path_obj.is_absolute():
path_obj = self.workspace_dir / path_obj
return path_obj.resolve()
def _is_safe_path(self, path: Path) -> bool:
"""Check if path is within allowed directories."""
try:
resolved = path.resolve()
for allowed in self.allowed_directories:
try:
resolved.relative_to(allowed.resolve())
return True
except ValueError:
continue
return False
except Exception:
return False
async def read_text_file(
self,
file_path: str,
encoding: str = "utf-8",
max_size_mb: int = 10
) -> Dict[str, Any]:
"""
Read a text file with size limits.
Args:
file_path: Path to the file
encoding: File encoding
max_size_mb: Maximum file size in MB
Returns:
Dictionary with file content and metadata
"""
try:
resolved_path = self._resolve_path(file_path)
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"Path {file_path} is outside allowed directories"
}
if not resolved_path.exists():
return {
"success": False,
"error": f"File {file_path} does not exist"
}
# Check file size
file_size = resolved_path.stat().st_size
max_size_bytes = max_size_mb * 1024 * 1024
if file_size > max_size_bytes:
return {
"success": False,
"error": f"File too large: {file_size / (1024*1024):.2f}MB (max: {max_size_mb}MB)"
}
# Read file
content = resolved_path.read_text(encoding=encoding)
return {
"success": True,
"content": content,
"file_path": str(resolved_path),
"file_size": file_size,
"encoding": encoding,
"lines": len(content.splitlines())
}
except UnicodeDecodeError:
return {
"success": False,
"error": f"File is not a valid text file with encoding {encoding}"
}
except Exception as e:
return {
"success": False,
"error": f"Failed to read file: {str(e)}"
}
async def read_multiple_files(
self,
file_paths: List[str],
encoding: str = "utf-8"
) -> Dict[str, Any]:
"""
Read multiple files at once.
Args:
file_paths: List of file paths
encoding: File encoding
Returns:
Dictionary with all file contents
"""
results = {}
errors = []
for file_path in file_paths:
result = await self.read_text_file(file_path, encoding)
if result["success"]:
results[file_path] = {
"content": result["content"],
"size": result["file_size"],
"lines": result["lines"]
}
else:
errors.append({
"file": file_path,
"error": result["error"]
})
return {
"success": len(results) > 0,
"files_read": len(results),
"files_failed": len(errors),
"results": results,
"errors": errors
}
async def list_directory_with_sizes(
self,
directory_path: str = "."
) -> Dict[str, Any]:
"""
List directory contents with file sizes.
Args:
directory_path: Path to directory
Returns:
Dictionary with directory contents and sizes
"""
try:
resolved_path = self._resolve_path(directory_path)
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"Path {directory_path} is outside allowed directories"
}
if not resolved_path.exists():
return {
"success": False,
"error": f"Directory {directory_path} does not exist"
}
if not resolved_path.is_dir():
return {
"success": False,
"error": f"{directory_path} is not a directory"
}
# List contents with sizes
contents = []
total_size = 0
for item in sorted(resolved_path.iterdir()):
try:
is_dir = item.is_dir()
size = 0 if is_dir else item.stat().st_size
total_size += size
contents.append({
"name": item.name,
"type": "directory" if is_dir else "file",
"size": size,
"size_human": self._format_size(size),
"modified": item.stat().st_mtime
})
except Exception as e:
contents.append({
"name": item.name,
"type": "unknown",
"error": str(e)
})
return {
"success": True,
"directory": str(resolved_path),
"total_items": len(contents),
"total_size": total_size,
"total_size_human": self._format_size(total_size),
"contents": contents
}
except Exception as e:
return {
"success": False,
"error": f"Failed to list directory: {str(e)}"
}
async def directory_tree(
self,
directory_path: str = ".",
max_depth: int = 3,
show_hidden: bool = False
) -> Dict[str, Any]:
"""
Generate a tree structure of directory contents.
Args:
directory_path: Path to directory
max_depth: Maximum depth to traverse
show_hidden: Whether to show hidden files
Returns:
Dictionary with directory tree structure
"""
try:
resolved_path = self._resolve_path(directory_path)
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"Path {directory_path} is outside allowed directories"
}
def build_tree(path: Path, current_depth: int = 0) -> Dict[str, Any]:
"""Recursively build tree structure."""
if current_depth >= max_depth:
return {"name": path.name, "type": "directory", "truncated": True}
if not path.is_dir():
return {
"name": path.name,
"type": "file",
"size": path.stat().st_size
}
children = []
try:
for item in sorted(path.iterdir()):
# Skip hidden files if needed
if not show_hidden and item.name.startswith('.'):
continue
children.append(build_tree(item, current_depth + 1))
except PermissionError:
return {
"name": path.name,
"type": "directory",
"error": "Permission denied"
}
return {
"name": path.name,
"type": "directory",
"children": children,
"count": len(children)
}
tree = build_tree(resolved_path)
return {
"success": True,
"root": str(resolved_path),
"tree": tree,
"max_depth": max_depth
}
except Exception as e:
return {
"success": False,
"error": f"Failed to generate directory tree: {str(e)}"
}
async def search_files(
self,
pattern: str,
directory_path: str = ".",
recursive: bool = True,
case_sensitive: bool = False
) -> Dict[str, Any]:
"""
Search for files matching a pattern.
Args:
pattern: Glob pattern (e.g., "*.py", "test_*.txt")
directory_path: Directory to search in
recursive: Search recursively
case_sensitive: Case-sensitive matching
Returns:
Dictionary with matching files
"""
try:
resolved_path = self._resolve_path(directory_path)
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"Path {directory_path} is outside allowed directories"
}
if not resolved_path.exists():
return {
"success": False,
"error": f"Directory {directory_path} does not exist"
}
# Search for files
if recursive:
matches = list(resolved_path.rglob(pattern))
else:
matches = list(resolved_path.glob(pattern))
# Filter to files only
files = [m for m in matches if m.is_file()]
results = []
for file_path in sorted(files):
try:
stat = file_path.stat()
results.append({
"path": str(file_path.relative_to(resolved_path)),
"absolute_path": str(file_path),
"size": stat.st_size,
"size_human": self._format_size(stat.st_size),
"modified": stat.st_mtime
})
except Exception:
pass
return {
"success": True,
"pattern": pattern,
"directory": str(resolved_path),
"recursive": recursive,
"matches": len(results),
"files": results
}
except Exception as e:
return {
"success": False,
"error": f"Failed to search files: {str(e)}"
}
async def get_file_info(
self,
file_path: str
) -> Dict[str, Any]:
"""
Get detailed information about a file.
Args:
file_path: Path to the file
Returns:
Dictionary with file information
"""
try:
resolved_path = self._resolve_path(file_path)
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"Path {file_path} is outside allowed directories"
}
if not resolved_path.exists():
return {
"success": False,
"error": f"File {file_path} does not exist"
}
stat = resolved_path.stat()
info = {
"path": str(resolved_path),
"name": resolved_path.name,
"extension": resolved_path.suffix,
"size": stat.st_size,
"size_human": self._format_size(stat.st_size),
"is_file": resolved_path.is_file(),
"is_directory": resolved_path.is_dir(),
"is_symlink": resolved_path.is_symlink(),
"created": stat.st_ctime,
"modified": stat.st_mtime,
"accessed": stat.st_atime,
"permissions": oct(stat.st_mode)[-3:]
}
# Add parent directory info
info["parent"] = str(resolved_path.parent)
# For text files, add line count
if resolved_path.is_file() and resolved_path.suffix in ['.txt', '.py', '.md', '.json', '.yaml', '.yml']:
try:
content = resolved_path.read_text()
info["lines"] = len(content.splitlines())
info["characters"] = len(content)
except Exception:
pass
return {
"success": True,
"file_info": info
}
except Exception as e:
return {
"success": False,
"error": f"Failed to get file info: {str(e)}"
}
async def move_file(
self,
source: str,
destination: str,
overwrite: bool = False
) -> Dict[str, Any]:
"""
Move or rename a file/directory.
Args:
source: Source path
destination: Destination path
overwrite: Whether to overwrite existing destination
Returns:
Dictionary with operation result
"""
try:
source_path = self._resolve_path(source)
dest_path = self._resolve_path(destination)
if not self._is_safe_path(source_path) or not self._is_safe_path(dest_path):
return {
"success": False,
"error": "Paths must be within allowed directories"
}
if not source_path.exists():
return {
"success": False,
"error": f"Source {source} does not exist"
}
if dest_path.exists() and not overwrite:
return {
"success": False,
"error": f"Destination {destination} already exists. Use overwrite=True to replace."
}
# Perform move
if dest_path.exists():
if dest_path.is_dir():
shutil.rmtree(dest_path)
else:
dest_path.unlink()
shutil.move(str(source_path), str(dest_path))
return {
"success": True,
"source": str(source_path),
"destination": str(dest_path),
"message": f"Moved {source} to {destination}"
}
except Exception as e:
return {
"success": False,
"error": f"Failed to move file: {str(e)}"
}
async def copy_file(
self,
source: str,
destination: str,
overwrite: bool = False
) -> Dict[str, Any]:
"""
Copy a file or directory.
Args:
source: Source path
destination: Destination path
overwrite: Whether to overwrite existing destination
Returns:
Dictionary with operation result
"""
try:
source_path = self._resolve_path(source)
dest_path = self._resolve_path(destination)
if not self._is_safe_path(source_path) or not self._is_safe_path(dest_path):
return {
"success": False,
"error": "Paths must be within allowed directories"
}
if not source_path.exists():
return {
"success": False,
"error": f"Source {source} does not exist"
}
if dest_path.exists() and not overwrite:
return {
"success": False,
"error": f"Destination {destination} already exists"
}
# Perform copy
if source_path.is_dir():
if dest_path.exists():
shutil.rmtree(dest_path)
shutil.copytree(source_path, dest_path)
else:
dest_path.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source_path, dest_path)
return {
"success": True,
"source": str(source_path),
"destination": str(dest_path),
"message": f"Copied {source} to {destination}"
}
except Exception as e:
return {
"success": False,
"error": f"Failed to copy file: {str(e)}"
}
async def delete_file(
self,
file_path: str,
recursive: bool = False
) -> Dict[str, Any]:
"""
Delete a file or directory.
Args:
file_path: Path to delete
recursive: For directories, delete recursively
Returns:
Dictionary with operation result
"""
try:
resolved_path = self._resolve_path(file_path)
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"Path {file_path} is outside allowed directories"
}
if not resolved_path.exists():
return {
"success": False,
"error": f"Path {file_path} does not exist"
}
# Delete
if resolved_path.is_dir():
if not recursive:
return {
"success": False,
"error": "Cannot delete directory without recursive=True"
}
shutil.rmtree(resolved_path)
else:
resolved_path.unlink()
return {
"success": True,
"deleted": str(resolved_path),
"message": f"Deleted {file_path}"
}
except Exception as e:
return {
"success": False,
"error": f"Failed to delete: {str(e)}"
}
async def create_directory(
self,
directory_path: str,
parents: bool = True
) -> Dict[str, Any]:
"""
Create a new directory.
Args:
directory_path: Path for new directory
parents: Create parent directories if needed
Returns:
Dictionary with operation result
"""
try:
resolved_path = self._resolve_path(directory_path)
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"Path {directory_path} is outside allowed directories"
}
if resolved_path.exists():
return {
"success": False,
"error": f"Directory {directory_path} already exists"
}
# Create directory
resolved_path.mkdir(parents=parents, exist_ok=False)
return {
"success": True,
"directory": str(resolved_path),
"message": f"Created directory {directory_path}"
}
except Exception as e:
return {
"success": False,
"error": f"Failed to create directory: {str(e)}"
}
async def list_allowed_directories(self) -> Dict[str, Any]:
"""
List directories that are accessible.
Returns:
Dictionary with allowed directories
"""
return {
"success": True,
"allowed_directories": [str(d) for d in self.allowed_directories],
"count": len(self.allowed_directories)
}
def _format_size(self, size_bytes: int) -> str:
"""Format file size in human-readable format."""
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
if size_bytes < 1024.0:
return f"{size_bytes:.2f} {unit}"
size_bytes /= 1024.0
return f"{size_bytes:.2f} PB"
hello.py¶
#!/usr/bin/env python3
"""A simple greeting script."""
def main():
name = "World"
print(f"Hello, {name}!")
if __name__ == "__main__":
main()
llm_helper.py¶
"""LLM helper for safety checks, approval, and summarization."""
import json
from typing import Optional, Dict, Any
from openai import OpenAI
from config import Config
def _reasoning_safe_temperature(model, requested=1.0):
"""Reasoning models (Kimi K3, GPT-5, ...) only accept temperature=1.
Return 1 for those; otherwise the requested value so non-reasoning
providers (Doubao, DeepSeek, older Moonshot) are unchanged."""
m = str(model or "").lower().replace("/", "-")
return 1 if ("kimi-k3" in m or "gpt-5" in m) else requested
def _parse_json_response(content):
"""Parse a JSON object out of an LLM reply, tolerating markdown fences.
Reasoning models (notably kimi-k3) reliably return valid JSON but wrap it
in a ```json ... ``` code fence, so a bare json.loads() fails with
"Expecting value: line 1 column 1". Strip an optional fence and, as a last
resort, slice from the first '{' to the last '}' before parsing."""
text = (content or "").strip()
if text.startswith("```"):
# Drop the opening fence line (``` or ```json) and the closing fence.
text = text.split("\n", 1)[1] if "\n" in text else ""
if text.rstrip().endswith("```"):
text = text.rstrip()[:-3]
text = text.strip()
try:
return json.loads(text)
except json.JSONDecodeError:
start, end = text.find("{"), text.rfind("}")
if start != -1 and end != -1 and end > start:
return json.loads(text[start:end + 1])
raise
class LLMHelper:
"""Helper class for LLM-based operations."""
def __init__(self):
"""Initialize the LLM helper.
The OpenAI-compatible client is created lazily on first use so that
execution tools which do not need an LLM (e.g. Python code execution
with local syntax checking, terminal commands, file writes) work
offline without any API key configured. Methods that actually call
the LLM (approval, summarization, non-Python syntax check) will raise
or fail-safe if no key is available.
"""
self.client = None
self.model = None
self.provider = None
def _ensure_client(self) -> None:
"""Create the LLM client on first use (raises if no API key)."""
if self.client is None:
llm_config = Config.get_llm_config()
# All providers use OpenAI-compatible API
self.client = OpenAI(
api_key=llm_config["api_key"],
base_url=llm_config.get("base_url")
)
self.model = llm_config["model"]
self.provider = llm_config["provider"]
def request_approval(
self,
operation: str,
details: Dict[str, Any]
) -> tuple[bool, str]:
"""
Request LLM approval for a dangerous operation.
Args:
operation: The operation name
details: Details about the operation
Returns:
Tuple of (approved, reason)
"""
prompt = f"""You are a safety reviewer for an AI agent execution system.
Review the following operation and determine if it should be approved.
Operation: {operation}
Details: {json.dumps(details, indent=2)}
Analyze the operation for:
1. Potential data loss or destructive actions
2. Security risks
3. Resource consumption concerns
4. Compliance with best practices
Respond in JSON format:
{{
"approved": true/false,
"reason": "Brief explanation of your decision",
"risk_level": "low/medium/high",
"recommendations": ["List of recommendations if any"]
}}
"""
try:
self._ensure_client()
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": "You are a cautious safety reviewer. Approve operations that are safe and reject risky ones."
},
{"role": "user", "content": prompt}
],
temperature=_reasoning_safe_temperature(self.model, 0.1),
max_tokens=Config.MAX_TOKENS
)
result = _parse_json_response(response.choices[0].message.content)
return result["approved"], result["reason"]
except Exception as e:
# If approval check fails, default to rejection for safety
return False, f"Approval check failed: {str(e)}"
def summarize_output(
self,
tool_name: str,
output: str
) -> str:
"""
Summarize complex tool output.
Args:
tool_name: Name of the tool that produced the output
output: The output to summarize
Returns:
Summarized output
"""
prompt = f"""Summarize the following output from the '{tool_name}' tool.
Focus on:
1. Key results or findings
2. Errors or warnings
3. Important patterns or insights
4. Actionable information
Output to summarize:
{output[:5000]} # Limit input to avoid token limits
Provide a concise summary that captures the essential information."""
try:
self._ensure_client()
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": "You are an expert at summarizing technical output. Be concise and focus on actionable information."
},
{"role": "user", "content": prompt}
],
temperature=_reasoning_safe_temperature(self.model, 0.1),
max_tokens=Config.MAX_TOKENS
)
summary = response.choices[0].message.content
return f"[SUMMARIZED OUTPUT]\n{summary}\n\n[Original output length: {len(output)} characters]"
except Exception as e:
return f"[SUMMARIZATION FAILED: {str(e)}]\n\n{output[:Config.MAX_OUTPUT_LENGTH]}..."
def analyze_error(
self,
tool_name: str,
command: str,
error_output: str
) -> str:
"""
Analyze error output and provide suggestions.
Args:
tool_name: Name of the tool that produced the error
command: The command or code that failed
error_output: The error output
Returns:
Analysis with suggestions
"""
prompt = f"""Analyze the following error from the '{tool_name}' tool:
Command/Code:
{command}
Error Output:
{error_output[:3000]}
Provide:
1. Root cause analysis
2. Suggested fixes
3. Prevention strategies
Be concise and practical."""
try:
self._ensure_client()
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": "You are an expert debugger. Analyze errors and provide clear, actionable solutions."
},
{"role": "user", "content": prompt}
],
temperature=_reasoning_safe_temperature(self.model, 0.2),
max_tokens=Config.MAX_TOKENS
)
return response.choices[0].message.content
except Exception as e:
return f"Error analysis failed: {str(e)}"
def verify_code_syntax(
self,
code: str,
language: str = "python"
) -> tuple[bool, Optional[str]]:
"""
Verify code syntax and provide feedback.
Args:
code: The code to verify
language: Programming language
Returns:
Tuple of (is_valid, error_message)
"""
# For Python, we can do actual syntax checking
if language == "python":
try:
compile(code, "<string>", "exec")
return True, None
except SyntaxError as e:
return False, f"Syntax error at line {e.lineno}: {e.msg}"
# For other languages, use LLM for basic validation
prompt = f"""Check the following {language} code for syntax errors:
```{language}
{code}
Respond in JSON format: {{ "valid": true/false, "errors": ["List of syntax errors if any"], "warnings": ["List of warnings if any"] }} """
try:
self._ensure_client()
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": f"You are a {language} syntax validator. Check code for syntax errors."
},
{"role": "user", "content": prompt}
],
temperature=_reasoning_safe_temperature(self.model, 0.1),
max_tokens=Config.MAX_TOKENS
)
result = _parse_json_response(response.choices[0].message.content)
if result["valid"]:
return True, None
else:
return False, "; ".join(result["errors"])
except Exception as e:
# If validation fails, allow the code through
return True, None
### `multilang_executor.py`
```python
"""Multi-language code execution support inspired by SandboxFusion."""
import asyncio
import subprocess
import tempfile
import os
import shutil
import time
import base64
import psutil
from typing import Dict, Any, Optional, List
from enum import Enum
import logging
logger = logging.getLogger(__name__)
def try_decode(s: bytes) -> str:
"""Safely decode bytes to string."""
try:
return s.decode('utf-8', errors='replace')
except Exception as e:
return f'[DecodeError] {e}'
async def get_all_output(stream) -> str:
"""Read stream until EOF. Call after the process has exited or been killed."""
if stream is None:
return ""
try:
result = await stream.read()
return try_decode(result)
except Exception as e:
logger.debug(f"Error reading output: {e}")
return ""
def kill_process_tree(pid: int):
"""Kill process and all its children."""
try:
parent = psutil.Process(pid)
children = parent.children(recursive=True)
# Kill children first
for child in children:
try:
child.kill()
except psutil.NoSuchProcess:
pass
# Kill parent
try:
parent.kill()
except psutil.NoSuchProcess:
pass
except psutil.NoSuchProcess:
pass
except Exception as e:
logger.warning(f'Error killing process tree: {e}')
class ExecutionStatus(str, Enum):
"""Execution status."""
SUCCESS = "success"
FAILED = "failed"
TIMEOUT = "timeout"
ERROR = "error"
class LanguageExecutor:
"""Multi-language code executor."""
def __init__(self, workspace_dir: str = None):
"""Initialize executor."""
self.workspace_dir = workspace_dir or os.getcwd()
async def execute_code(
self,
code: str,
language: str,
timeout: float = 30.0,
compile_timeout: float = 10.0,
stdin: Optional[str] = None,
files: Optional[Dict[str, str]] = None
) -> Dict[str, Any]:
"""
Execute code in the specified language.
Args:
code: Code to execute
language: Programming language
timeout: Execution timeout in seconds
compile_timeout: Compilation timeout in seconds
stdin: Optional stdin input
files: Optional additional files (name -> content)
Returns:
Execution result dictionary
"""
if language is None:
language = "python"
language = language.lower()
# Map language to executor
executors = {
'python': self._run_python,
'python3': self._run_python,
'javascript': self._run_javascript,
'js': self._run_javascript,
'typescript': self._run_typescript,
'ts': self._run_typescript,
'go': self._run_go,
'java': self._run_java,
'cpp': self._run_cpp,
'c++': self._run_cpp,
'rust': self._run_rust,
'php': self._run_php,
'bash': self._run_bash,
'shell': self._run_bash,
'sh': self._run_bash,
'nodejs': self._run_javascript,
'node': self._run_javascript,
}
executor = executors.get(language)
if not executor:
return {
"status": ExecutionStatus.ERROR,
"error": f"Unsupported language: {language}. Supported: {', '.join(sorted(set(executors.keys())))}"
}
try:
return await executor(code, timeout, compile_timeout, stdin, files or {})
except Exception as e:
logger.exception(f"Error executing {language} code")
return {
"status": ExecutionStatus.ERROR,
"error": f"Execution failed: {str(e)}"
}
async def _run_command(
self,
command: str,
timeout: float,
stdin: Optional[str] = None,
cwd: Optional[str] = None,
shell: bool = True
) -> Dict[str, Any]:
"""Run a shell command and return results with proper process management."""
process = None
try:
logger.debug(f'Running command: {command[:100]}...')
process = await asyncio.create_subprocess_shell(
command,
stdin=asyncio.subprocess.PIPE if stdin else None,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=cwd,
executable='/bin/bash'
)
# Write stdin if provided
if stdin and process.stdin:
try:
process.stdin.write(stdin.encode())
await process.stdin.drain()
process.stdin.close()
except Exception as e:
logger.warning(f"Failed to write stdin: {e}")
start_time = time.time()
try:
# Wait for process with timeout
await asyncio.wait_for(process.wait(), timeout=timeout)
execution_time = time.time() - start_time
# Read output non-blocking
stdout = await get_all_output(process.stdout)
stderr = await get_all_output(process.stderr)
logger.debug(f'Command completed in {execution_time:.2f}s')
return {
"status": ExecutionStatus.SUCCESS if process.returncode == 0 else ExecutionStatus.FAILED,
"returncode": process.returncode,
"stdout": stdout,
"stderr": stderr,
"execution_time": execution_time
}
except asyncio.TimeoutError:
execution_time = time.time() - start_time
# Kill first so pipes close, then drain remaining output
if psutil.pid_exists(process.pid):
kill_process_tree(process.pid)
logger.info(f'Process {process.pid} killed due to timeout')
stdout = await get_all_output(process.stdout)
stderr = await get_all_output(process.stderr)
return {
"status": ExecutionStatus.TIMEOUT,
"error": f"Execution timed out after {timeout} seconds",
"stdout": stdout,
"stderr": stderr,
"execution_time": execution_time
}
except Exception as e:
logger.exception(f"Error running command: {command[:100]}")
return {
"status": ExecutionStatus.ERROR,
"error": f"Command execution failed: {str(e)}"
}
finally:
# Cleanup: ensure process is terminated
if process and psutil.pid_exists(process.pid):
kill_process_tree(process.pid)
def _write_files(self, tmp_dir: str, files: Dict[str, str]):
"""Write additional files to tmp directory."""
for filename, content in files.items():
if not content or "IGNORE_THIS_FILE" in filename:
continue
filepath = os.path.join(tmp_dir, filename)
dirpath = os.path.dirname(filepath)
if dirpath:
os.makedirs(dirpath, exist_ok=True)
# Handle base64 encoded content
try:
if self._is_base64(content):
with open(filepath, 'wb') as f:
f.write(base64.b64decode(content))
else:
with open(filepath, 'w', encoding='utf-8') as f:
f.write(content)
except Exception as e:
logger.warning(f"Failed to write file {filename}: {e}")
def _is_base64(self, s: str) -> bool:
"""Check if string is base64 encoded."""
try:
if len(s) % 4 != 0 or not all(c in 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=' for c in s):
return False
base64.b64decode(s, validate=True)
return True
except Exception:
return False
async def _run_python(
self,
code: str,
timeout: float,
compile_timeout: float,
stdin: Optional[str],
files: Dict[str, str]
) -> Dict[str, Any]:
"""Execute Python code."""
with tempfile.TemporaryDirectory(prefix='python_', ignore_cleanup_errors=True) as tmp_dir:
self._write_files(tmp_dir, files)
code_file = os.path.join(tmp_dir, 'main.py')
with open(code_file, 'w', encoding='utf-8') as f:
f.write(code)
# Use python3 with unbuffered output
result = await self._run_command(
f'python3 -u {code_file}',
timeout,
stdin,
tmp_dir
)
result['language'] = 'python'
return result
async def _run_javascript(
self,
code: str,
timeout: float,
compile_timeout: float,
stdin: Optional[str],
files: Dict[str, str]
) -> Dict[str, Any]:
"""Execute JavaScript code with Node.js."""
with tempfile.TemporaryDirectory(prefix='js_', ignore_cleanup_errors=True) as tmp_dir:
self._write_files(tmp_dir, files)
# Create package.json if not exists to enable ES modules
if 'package.json' not in files:
package_json = {
"type": "module",
"dependencies": {}
}
with open(os.path.join(tmp_dir, 'package.json'), 'w') as f:
import json
json.dump(package_json, f)
code_file = os.path.join(tmp_dir, 'main.js')
with open(code_file, 'w', encoding='utf-8') as f:
f.write(code)
result = await self._run_command(
f'node {code_file}',
timeout,
stdin,
tmp_dir
)
result['language'] = 'javascript'
return result
async def _run_typescript(
self,
code: str,
timeout: float,
compile_timeout: float,
stdin: Optional[str],
files: Dict[str, str]
) -> Dict[str, Any]:
"""Execute TypeScript code with tsx."""
with tempfile.TemporaryDirectory(prefix='ts_', ignore_cleanup_errors=True) as tmp_dir:
self._write_files(tmp_dir, files)
code_file = os.path.join(tmp_dir, 'main.ts')
with open(code_file, 'w', encoding='utf-8') as f:
f.write(code)
# Check if tsx is available, fallback to ts-node
check_tsx = await self._run_command('which tsx 2>/dev/null', 1.0)
cmd = 'tsx' if check_tsx['status'] == ExecutionStatus.SUCCESS else 'ts-node'
result = await self._run_command(
f'{cmd} {code_file}',
timeout,
stdin,
tmp_dir
)
result['language'] = 'typescript'
return result
async def _run_go(
self,
code: str,
timeout: float,
compile_timeout: float,
stdin: Optional[str],
files: Dict[str, str]
) -> Dict[str, Any]:
"""Execute Go code."""
with tempfile.TemporaryDirectory(prefix='go_', ignore_cleanup_errors=True) as tmp_dir:
self._write_files(tmp_dir, files)
# Initialize go module (ignore errors if already exists)
await self._run_command('go mod init main 2>/dev/null || true', 2.0, cwd=tmp_dir)
code_file = os.path.join(tmp_dir, 'main.go')
with open(code_file, 'w', encoding='utf-8') as f:
f.write(code)
# Compile
compile_result = await self._run_command(
'go build -o main main.go',
compile_timeout,
cwd=tmp_dir
)
if compile_result['status'] != ExecutionStatus.SUCCESS:
return {
"status": ExecutionStatus.FAILED,
"language": "go",
"phase": "compilation",
"returncode": compile_result.get('returncode', 1),
"stdout": compile_result.get('stdout', ''),
"stderr": compile_result.get('stderr', ''),
"error": "Compilation failed"
}
# Run
result = await self._run_command(
'./main',
timeout,
stdin,
tmp_dir
)
result['language'] = 'go'
result['compile_stdout'] = compile_result.get('stdout', '')
result['compile_stderr'] = compile_result.get('stderr', '')
return result
async def _run_java(
self,
code: str,
timeout: float,
compile_timeout: float,
stdin: Optional[str],
files: Dict[str, str]
) -> Dict[str, Any]:
"""Execute Java code."""
with tempfile.TemporaryDirectory(prefix='java_', ignore_cleanup_errors=True) as tmp_dir:
self._write_files(tmp_dir, files)
# Extract class name from public class declaration
class_name = 'Main'
import re
match = re.search(r'public\s+class\s+(\w+)', code)
if match:
class_name = match.group(1)
code_file = os.path.join(tmp_dir, f'{class_name}.java')
with open(code_file, 'w', encoding='utf-8') as f:
f.write(code)
# Prepare classpath for additional jars
jars = [f for f in files.keys() if f.endswith('.jar')]
classpath = '.:' + ':'.join(jars) if jars else '.'
# Compile
compile_result = await self._run_command(
f'javac -cp {classpath} {class_name}.java',
compile_timeout,
cwd=tmp_dir
)
if compile_result['status'] != ExecutionStatus.SUCCESS:
return {
"status": ExecutionStatus.FAILED,
"language": "java",
"phase": "compilation",
"returncode": compile_result.get('returncode', 1),
"stdout": compile_result.get('stdout', ''),
"stderr": compile_result.get('stderr', ''),
"error": "Compilation failed"
}
# Run with assertions enabled
result = await self._run_command(
f'java -cp {classpath} -ea {class_name}',
timeout,
stdin,
tmp_dir
)
result['language'] = 'java'
result['compile_stdout'] = compile_result.get('stdout', '')
result['compile_stderr'] = compile_result.get('stderr', '')
return result
async def _run_cpp(
self,
code: str,
timeout: float,
compile_timeout: float,
stdin: Optional[str],
files: Dict[str, str]
) -> Dict[str, Any]:
"""Execute C++ code."""
with tempfile.TemporaryDirectory(prefix='cpp_', ignore_cleanup_errors=True) as tmp_dir:
self._write_files(tmp_dir, files)
code_file = os.path.join(tmp_dir, 'main.cpp')
with open(code_file, 'w', encoding='utf-8') as f:
f.write(code)
# Compile with commonly needed flags
# Try with optional libraries (crypto, ssl, pthread)
compile_flags = '-std=c++17 -O2'
optional_libs = []
# Check if we need pthread
if '#include <thread>' in code or 'std::thread' in code:
optional_libs.append('-lpthread')
libs = ' '.join(optional_libs)
compile_result = await self._run_command(
f'g++ {compile_flags} main.cpp -o main {libs}',
compile_timeout,
cwd=tmp_dir
)
if compile_result['status'] != ExecutionStatus.SUCCESS:
return {
"status": ExecutionStatus.FAILED,
"language": "cpp",
"phase": "compilation",
"returncode": compile_result.get('returncode', 1),
"stdout": compile_result.get('stdout', ''),
"stderr": compile_result.get('stderr', ''),
"error": "Compilation failed"
}
# Run
result = await self._run_command(
'./main',
timeout,
stdin,
tmp_dir
)
result['language'] = 'cpp'
result['compile_stdout'] = compile_result.get('stdout', '')
result['compile_stderr'] = compile_result.get('stderr', '')
return result
async def _run_rust(
self,
code: str,
timeout: float,
compile_timeout: float,
stdin: Optional[str],
files: Dict[str, str]
) -> Dict[str, Any]:
"""Execute Rust code."""
with tempfile.TemporaryDirectory(prefix='rust_', ignore_cleanup_errors=True) as tmp_dir:
self._write_files(tmp_dir, files)
code_file = os.path.join(tmp_dir, 'main.rs')
with open(code_file, 'w', encoding='utf-8') as f:
f.write(code)
# Compile with optimizations
compile_result = await self._run_command(
'rustc -O main.rs -o main',
compile_timeout,
cwd=tmp_dir
)
if compile_result['status'] != ExecutionStatus.SUCCESS:
return {
"status": ExecutionStatus.FAILED,
"language": "rust",
"phase": "compilation",
"returncode": compile_result.get('returncode', 1),
"stdout": compile_result.get('stdout', ''),
"stderr": compile_result.get('stderr', ''),
"error": "Compilation failed"
}
# Run
result = await self._run_command(
'./main',
timeout,
stdin,
tmp_dir
)
result['language'] = 'rust'
result['compile_stdout'] = compile_result.get('stdout', '')
result['compile_stderr'] = compile_result.get('stderr', '')
return result
async def _run_php(
self,
code: str,
timeout: float,
compile_timeout: float,
stdin: Optional[str],
files: Dict[str, str]
) -> Dict[str, Any]:
"""Execute PHP code."""
with tempfile.TemporaryDirectory(prefix='php_', ignore_cleanup_errors=True) as tmp_dir:
self._write_files(tmp_dir, files)
# Ensure PHP tags
code_clean = code.strip()
if not code_clean.startswith('<?php') and not code_clean.startswith('<?'):
code = '<?php\n' + code
code_file = os.path.join(tmp_dir, 'main.php')
with open(code_file, 'w', encoding='utf-8') as f:
f.write(code)
result = await self._run_command(
f'php -f {code_file}',
timeout,
stdin,
tmp_dir
)
result['language'] = 'php'
return result
async def _run_bash(
self,
code: str,
timeout: float,
compile_timeout: float,
stdin: Optional[str],
files: Dict[str, str]
) -> Dict[str, Any]:
"""Execute Bash script."""
with tempfile.TemporaryDirectory(prefix='bash_', ignore_cleanup_errors=True) as tmp_dir:
self._write_files(tmp_dir, files)
code_file = os.path.join(tmp_dir, 'script.sh')
with open(code_file, 'w', encoding='utf-8') as f:
# Add shebang if not present
if not code.startswith('#!'):
f.write('#!/bin/bash\n')
f.write(code)
# Make executable
os.chmod(code_file, 0o755)
result = await self._run_command(
f'bash {code_file}',
timeout,
stdin,
tmp_dir
)
result['language'] = 'bash'
return result
quickstart.py¶
"""Quick start guide for the execution tools MCP server."""
import asyncio
from llm_helper import LLMHelper
from file_tools import FileTools
from execution_tools import ExecutionTools
async def quickstart():
"""Quick demonstration of the execution tools."""
print("=== Execution Tools MCP Server - Quick Start ===\n")
# Initialize
print("Initializing tools...")
llm_helper = LLMHelper()
file_tools = FileTools(llm_helper)
execution_tools = ExecutionTools(llm_helper)
# 1. File operations
print("\n1. File Operations Demo")
print("-" * 50)
print("\nWriting a Python script...")
result = await file_tools.write_file(
path="hello.py",
content="""#!/usr/bin/env python3
\"\"\"A simple greeting script.\"\"\"
def main():
name = "World"
print(f"Hello, {name}!")
if __name__ == "__main__":
main()
""",
overwrite=True
)
print(f"Status: {'✓ Success' if result['success'] else '✗ Failed'}")
if result['success']:
print(f"Written to: {result['path']}")
print(f"Verification: {result['verification']}")
# 2. Code execution
print("\n2. Code Interpreter Demo")
print("-" * 50)
print("\nExecuting Python code...")
result = await execution_tools.code_interpreter(
code="""
# Calculate fibonacci sequence
def fibonacci(n):
if n <= 1:
return n
return fibonacci(n-1) + fibonacci(n-2)
print("Fibonacci sequence (first 10 numbers):")
for i in range(10):
print(f"F({i}) = {fibonacci(i)}")
"""
)
print(f"Status: {'✓ Success' if result['success'] else '✗ Failed'}")
if result['success']:
print("Output:")
print(result['stdout'][:500]) # Print first 500 chars
# 3. Terminal execution
print("\n3. Virtual Terminal Demo")
print("-" * 50)
print("\nExecuting shell command...")
result = await execution_tools.virtual_terminal(
command="python --version && echo 'Current directory:' && pwd"
)
print(f"Status: {'✓ Success' if result['success'] else '✗ Failed'}")
if result['success']:
print("Output:")
print(result['stdout'])
# Summary
print("\n" + "=" * 50)
print("Quick start completed!")
print("\nKey Features:")
print(" • File operations with automatic syntax verification")
print(" • Code execution with error analysis")
print(" • Shell commands with result summarization")
print(" • LLM-based approval for dangerous operations")
print(" • External integrations (Google Calendar, GitHub)")
print("\nNext Steps:")
print(" • Run 'python examples.py' for more examples")
print(" • Run 'python server.py' to start the MCP server")
print(" • See README.md for full documentation")
if __name__ == "__main__":
asyncio.run(quickstart())
server.py¶
"""MCP server for execution tools."""
import asyncio
import json
from typing import Any
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
from config import Config
from llm_helper import LLMHelper
from file_tools import FileTools
from execution_tools import ExecutionTools
from external_tools import ExternalTools
# Initialize server
server = Server("execution-tools")
# Initialize tools
llm_helper = LLMHelper()
file_tools = FileTools(llm_helper)
execution_tools = ExecutionTools(llm_helper)
external_tools = ExternalTools(llm_helper)
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""List available tools."""
return [
types.Tool(
name="file_write",
description="Write content to a file with automatic syntax verification",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "File path (relative to workspace or absolute)"
},
"content": {
"type": "string",
"description": "Content to write"
},
"overwrite": {
"type": "boolean",
"description": "Whether to overwrite existing files",
"default": False
}
},
"required": ["path", "content"]
}
),
types.Tool(
name="file_edit",
description="Edit an existing file by searching and replacing content",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "File path"
},
"search": {
"type": "string",
"description": "Text to search for"
},
"replace": {
"type": "string",
"description": "Replacement text"
}
},
"required": ["path", "search", "replace"]
}
),
types.Tool(
name="code_interpreter",
description="Execute code in multiple programming languages in a sandboxed environment with result analysis. Supports: Python, JavaScript, TypeScript, Go, Java, C++, Rust, PHP, Bash",
inputSchema={
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "Code to execute"
},
"language": {
"type": "string",
"description": "Programming language (python, javascript, typescript, go, java, cpp, rust, php, bash)",
"default": "python"
},
"timeout": {
"type": "number",
"description": "Execution timeout in seconds",
"default": 30.0
},
"stdin": {
"type": "string",
"description": "Optional stdin input for the program"
},
"files": {
"type": "object",
"description": "Optional additional files (filename -> content mapping)",
"additionalProperties": {"type": "string"}
}
},
"required": ["code"]
}
),
types.Tool(
name="virtual_terminal",
description="Execute shell commands with error summarization",
inputSchema={
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "Shell command to execute"
},
"timeout": {
"type": "integer",
"description": "Timeout in seconds",
"default": 30
}
},
"required": ["command"]
}
),
types.Tool(
name="google_calendar_add",
description="Add an event to Google Calendar",
inputSchema={
"type": "object",
"properties": {
"summary": {
"type": "string",
"description": "Event title"
},
"start_time": {
"type": "string",
"description": "Start time (ISO 8601 format, e.g., 2024-01-01T10:00:00)"
},
"end_time": {
"type": "string",
"description": "End time (ISO 8601 format)"
},
"description": {
"type": "string",
"description": "Event description"
},
"location": {
"type": "string",
"description": "Event location"
}
},
"required": ["summary", "start_time", "end_time"]
}
),
types.Tool(
name="github_create_pr",
description="Create a GitHub Pull Request",
inputSchema={
"type": "object",
"properties": {
"repo_name": {
"type": "string",
"description": "Repository name (format: owner/repo)"
},
"title": {
"type": "string",
"description": "PR title"
},
"body": {
"type": "string",
"description": "PR description"
},
"head_branch": {
"type": "string",
"description": "Source branch"
},
"base_branch": {
"type": "string",
"description": "Target branch",
"default": "main"
}
},
"required": ["repo_name", "title", "body", "head_branch"]
}
)
]
@server.call_tool()
async def handle_call_tool(
name: str,
arguments: dict[str, Any] | None
) -> list[types.TextContent]:
"""Handle tool calls."""
if arguments is None:
arguments = {}
try:
# Route to appropriate tool
if name == "file_write":
result = await file_tools.write_file(
path=arguments["path"],
content=arguments["content"],
overwrite=arguments.get("overwrite", False)
)
elif name == "file_edit":
result = await file_tools.edit_file(
path=arguments["path"],
search=arguments["search"],
replace=arguments["replace"]
)
elif name == "code_interpreter":
result = await execution_tools.code_interpreter(
code=arguments["code"],
language=arguments.get("language") or "python",
timeout=arguments.get("timeout", 30.0),
stdin=arguments.get("stdin"),
files=arguments.get("files")
)
elif name == "virtual_terminal":
result = await execution_tools.virtual_terminal(
command=arguments["command"],
timeout=arguments.get("timeout", 30)
)
elif name == "google_calendar_add":
result = await external_tools.google_calendar_add(
summary=arguments["summary"],
start_time=arguments["start_time"],
end_time=arguments["end_time"],
description=arguments.get("description"),
location=arguments.get("location")
)
elif name == "github_create_pr":
result = await external_tools.github_create_pr(
repo_name=arguments["repo_name"],
title=arguments["title"],
body=arguments["body"],
head_branch=arguments["head_branch"],
base_branch=arguments.get("base_branch", "main")
)
else:
raise ValueError(f"Unknown tool: {name}")
# Format result
return [
types.TextContent(
type="text",
text=json.dumps(result, indent=2)
)
]
except Exception as e:
return [
types.TextContent(
type="text",
text=json.dumps({
"success": False,
"error": f"Tool execution failed: {str(e)}"
}, indent=2)
)
]
async def main():
"""Run the MCP server."""
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="execution-tools",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={}
)
)
)
if __name__ == "__main__":
asyncio.run(main())
terminal_controller.py¶
"""
Terminal Controller with integrated file operations.
Based on AWorld terminal-controller implementation.
Provides command execution with directory navigation and file editing.
"""
import os
import subprocess
import traceback
from pathlib import Path
from typing import Dict, Any
from config import Config
class TerminalController:
"""Terminal controller with directory navigation and file operations."""
def __init__(self):
self.workspace_dir = Path(Config.WORKSPACE_DIR).resolve()
self.current_directory = self.workspace_dir
self.command_history = []
self.max_history = 100
def _is_safe_path(self, path: Path) -> bool:
"""Check if path is within workspace."""
try:
path.resolve().relative_to(self.workspace_dir)
return True
except ValueError:
return False
def _resolve_path(self, path: str) -> Path:
"""Resolve path relative to current directory."""
path_obj = Path(path)
if not path_obj.is_absolute():
path_obj = self.current_directory / path_obj
return path_obj.resolve()
async def execute_command(
self,
command: str,
timeout: int = 30
) -> Dict[str, Any]:
"""
Execute a shell command in current directory.
Args:
command: Command to execute
timeout: Timeout in seconds
Returns:
Dictionary with command output
"""
try:
# Add to history
self.command_history.append(command)
if len(self.command_history) > self.max_history:
self.command_history.pop(0)
# Execute command
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=timeout,
cwd=str(self.current_directory)
)
return {
"success": result.returncode == 0,
"command": command,
"stdout": result.stdout,
"stderr": result.stderr,
"returncode": result.returncode,
"cwd": str(self.current_directory)
}
except subprocess.TimeoutExpired:
return {
"success": False,
"error": f"Command timed out after {timeout} seconds",
"command": command
}
except Exception as e:
return {
"success": False,
"error": f"Command execution failed: {str(e)}",
"command": command
}
async def get_current_directory(self) -> Dict[str, Any]:
"""
Get the current working directory.
Returns:
Dictionary with current directory path
"""
return {
"success": True,
"current_directory": str(self.current_directory),
"workspace": str(self.workspace_dir)
}
async def change_directory(
self,
directory: str
) -> Dict[str, Any]:
"""
Change the current working directory.
Args:
directory: Directory to change to
Returns:
Dictionary with new directory
"""
try:
new_dir = self._resolve_path(directory)
if not self._is_safe_path(new_dir):
return {
"success": False,
"error": f"Directory {directory} is outside workspace"
}
if not new_dir.exists():
return {
"success": False,
"error": f"Directory {directory} does not exist"
}
if not new_dir.is_dir():
return {
"success": False,
"error": f"{directory} is not a directory"
}
self.current_directory = new_dir
return {
"success": True,
"current_directory": str(self.current_directory),
"message": f"Changed to {directory}"
}
except Exception as e:
return {
"success": False,
"error": f"Failed to change directory: {str(e)}"
}
async def list_directory(
self,
directory: str = "."
) -> Dict[str, Any]:
"""
List contents of a directory.
Args:
directory: Directory to list (relative to current)
Returns:
Dictionary with directory contents
"""
try:
dir_path = self._resolve_path(directory)
if not self._is_safe_path(dir_path):
return {
"success": False,
"error": f"Directory {directory} is outside workspace"
}
if not dir_path.exists():
return {
"success": False,
"error": f"Directory {directory} does not exist"
}
contents = []
for item in sorted(dir_path.iterdir()):
contents.append({
"name": item.name,
"type": "directory" if item.is_dir() else "file",
"size": 0 if item.is_dir() else item.stat().st_size
})
return {
"success": True,
"directory": str(dir_path),
"contents": contents,
"count": len(contents)
}
except Exception as e:
return {
"success": False,
"error": f"Failed to list directory: {str(e)}"
}
async def read_file(
self,
file_path: str,
encoding: str = "utf-8"
) -> Dict[str, Any]:
"""
Read a file from current directory.
Args:
file_path: File path relative to current directory
encoding: File encoding
Returns:
Dictionary with file content
"""
try:
resolved_path = self._resolve_path(file_path)
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"File {file_path} is outside workspace"
}
if not resolved_path.exists():
return {
"success": False,
"error": f"File {file_path} does not exist"
}
content = resolved_path.read_text(encoding=encoding)
return {
"success": True,
"file_path": str(resolved_path),
"content": content,
"size": len(content),
"lines": len(content.splitlines())
}
except Exception as e:
return {
"success": False,
"error": f"Failed to read file: {str(e)}"
}
async def write_file(
self,
file_path: str,
content: str,
mode: str = "w"
) -> Dict[str, Any]:
"""
Write content to a file.
Args:
file_path: File path relative to current directory
content: Content to write
mode: Write mode ('w' for write, 'a' for append)
Returns:
Dictionary with write result
"""
try:
resolved_path = self._resolve_path(file_path)
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"File {file_path} is outside workspace"
}
# Create parent directories if needed
resolved_path.parent.mkdir(parents=True, exist_ok=True)
# Write file
if mode == "a":
with open(resolved_path, 'a', encoding='utf-8') as f:
f.write(content)
else:
resolved_path.write_text(content, encoding='utf-8')
return {
"success": True,
"file_path": str(resolved_path),
"bytes_written": len(content),
"mode": mode
}
except Exception as e:
return {
"success": False,
"error": f"Failed to write file: {str(e)}"
}
async def insert_file_content(
self,
file_path: str,
content: str,
line_number: int
) -> Dict[str, Any]:
"""
Insert content at specific line in file.
Args:
file_path: File path
content: Content to insert
line_number: Line number to insert at (1-indexed)
Returns:
Dictionary with operation result
"""
try:
resolved_path = self._resolve_path(file_path)
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"File {file_path} is outside workspace"
}
if not resolved_path.exists():
return {
"success": False,
"error": f"File {file_path} does not exist"
}
# Read current content
lines = resolved_path.read_text().splitlines()
# Insert content
if line_number < 1 or line_number > len(lines) + 1:
return {
"success": False,
"error": f"Line number {line_number} out of range (1-{len(lines)+1})"
}
lines.insert(line_number - 1, content)
# Write back
resolved_path.write_text('\n'.join(lines) + '\n')
return {
"success": True,
"file_path": str(resolved_path),
"line_number": line_number,
"total_lines": len(lines)
}
except Exception as e:
return {
"success": False,
"error": f"Failed to insert content: {str(e)}"
}
async def delete_file_content(
self,
file_path: str,
start_line: int,
end_line: int
) -> Dict[str, Any]:
"""
Delete lines from file.
Args:
file_path: File path
start_line: Start line number (1-indexed, inclusive)
end_line: End line number (1-indexed, inclusive)
Returns:
Dictionary with operation result
"""
try:
resolved_path = self._resolve_path(file_path)
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"File {file_path} is outside workspace"
}
if not resolved_path.exists():
return {
"success": False,
"error": f"File {file_path} does not exist"
}
# Read lines
lines = resolved_path.read_text().splitlines()
# Validate range
if start_line < 1 or end_line > len(lines) or start_line > end_line:
return {
"success": False,
"error": f"Invalid line range: {start_line}-{end_line} (file has {len(lines)} lines)"
}
# Delete lines
del lines[start_line - 1:end_line]
# Write back
resolved_path.write_text('\n'.join(lines) + '\n' if lines else '')
return {
"success": True,
"file_path": str(resolved_path),
"deleted_lines": end_line - start_line + 1,
"remaining_lines": len(lines)
}
except Exception as e:
return {
"success": False,
"error": f"Failed to delete content: {str(e)}"
}
async def update_file_content(
self,
file_path: str,
line_number: int,
new_content: str
) -> Dict[str, Any]:
"""
Update a specific line in file.
Args:
file_path: File path
line_number: Line number to update (1-indexed)
new_content: New content for the line
Returns:
Dictionary with operation result
"""
try:
resolved_path = self._resolve_path(file_path)
if not self._is_safe_path(resolved_path):
return {
"success": False,
"error": f"File {file_path} is outside workspace"
}
if not resolved_path.exists():
return {
"success": False,
"error": f"File {file_path} does not exist"
}
# Read lines
lines = resolved_path.read_text().splitlines()
if line_number < 1 or line_number > len(lines):
return {
"success": False,
"error": f"Line number {line_number} out of range (1-{len(lines)})"
}
# Update line
old_content = lines[line_number - 1]
lines[line_number - 1] = new_content
# Write back
resolved_path.write_text('\n'.join(lines) + '\n')
return {
"success": True,
"file_path": str(resolved_path),
"line_number": line_number,
"old_content": old_content,
"new_content": new_content
}
except Exception as e:
return {
"success": False,
"error": f"Failed to update content: {str(e)}"
}
async def get_command_history(
self,
count: int = 10
) -> Dict[str, Any]:
"""
Get recent command history.
Args:
count: Number of recent commands
Returns:
Dictionary with command history
"""
recent = self.command_history[-count:] if self.command_history else []
return {
"success": True,
"history": recent,
"count": len(recent),
"total": len(self.command_history)
}
test_config_env.py¶
"""Regression test: malformed numeric env vars must not crash config import.
TEMPERATURE / MAX_TOKENS / MAX_OUTPUT_LENGTH were parsed with bare
float()/int() at import time, so e.g. MAX_TOKENS=abc crashed every tool with
ValueError. They now fall back to defaults with a warning.
"""
import importlib
import os
import sys
sys.path.insert(0, os.path.dirname(__file__))
# Ensure the real local config module is imported (other tests stub it).
sys.modules.pop("config", None)
import config as cfg
def test_env_int_falls_back_on_malformed(monkeypatch, capsys):
monkeypatch.setenv("MAX_TOKENS", "abc")
assert cfg._env_int("MAX_TOKENS", 4096) == 4096
assert "invalid MAX_TOKENS" in capsys.readouterr().err
def test_env_int_parses_valid_value(monkeypatch):
monkeypatch.setenv("MAX_TOKENS", "123")
assert cfg._env_int("MAX_TOKENS", 4096) == 123
def test_env_float_falls_back_on_malformed(monkeypatch, capsys):
monkeypatch.setenv("TEMPERATURE", "hot")
assert cfg._env_float("TEMPERATURE", 0.7) == 0.7
assert "invalid TEMPERATURE" in capsys.readouterr().err
def test_env_float_parses_valid_value(monkeypatch):
monkeypatch.setenv("TEMPERATURE", "0.2")
assert cfg._env_float("TEMPERATURE", 0.7) == 0.2
def test_module_import_survives_malformed_env(monkeypatch):
"""Import-time class attributes must not raise on malformed env values."""
monkeypatch.setenv("MAX_OUTPUT_LENGTH", "lots")
# Fresh import from disk (test_terminal_controller stubs sys.modules['config']).
sys.modules.pop("config", None)
fresh = importlib.import_module("config")
assert fresh.Config.MAX_OUTPUT_LENGTH == 1000
test_execution_tools.py¶
"""Test execution tools."""
import asyncio
from llm_helper import LLMHelper
from execution_tools import ExecutionTools
async def test_code_interpreter():
"""Test code interpreter functionality."""
print("Testing code interpreter...")
llm_helper = LLMHelper()
execution_tools = ExecutionTools(llm_helper)
# Test valid code
result = await execution_tools.code_interpreter(
code='print("Test successful")\nresult = 2 + 2\nprint(f"2 + 2 = {result}")'
)
assert result["success"], f"Code execution failed: {result.get('error')}"
assert "Test successful" in result["stdout"]
print(f"✓ Code execution successful: {result}")
# Test error handling
result = await execution_tools.code_interpreter(
code='x = 1 / 0'
)
assert not result["success"], "Should fail with division by zero"
assert "error_analysis" in result
print(f"✓ Error handling works: {result['error'][:100]}...")
async def test_virtual_terminal():
"""Test virtual terminal functionality."""
print("\nTesting virtual terminal...")
llm_helper = LLMHelper()
execution_tools = ExecutionTools(llm_helper)
# Test successful command
result = await execution_tools.virtual_terminal(
command='echo "Terminal test"'
)
assert result["success"], f"Command failed: {result.get('error')}"
assert "Terminal test" in result["stdout"]
print(f"✓ Command execution successful: {result}")
# Test failed command
result = await execution_tools.virtual_terminal(
command='ls /nonexistent_directory_12345'
)
assert not result["success"], "Should fail with non-existent directory"
assert "error_analysis" in result
print(f"✓ Error handling works: returncode={result['returncode']}")
async def test_syntax_verification():
"""Test syntax verification."""
print("\nTesting syntax verification...")
llm_helper = LLMHelper()
execution_tools = ExecutionTools(llm_helper)
# Test syntax error detection
result = await execution_tools.code_interpreter(
code='print("Unclosed string'
)
assert not result["success"], "Should detect syntax error"
print(f"✓ Syntax verification works: {result['error'][:100]}...")
async def main():
"""Run all tests."""
print("=== Execution Tools Tests ===\n")
try:
await test_code_interpreter()
await test_virtual_terminal()
await test_syntax_verification()
print("\n✓ All execution tools tests passed!")
except AssertionError as e:
print(f"\n✗ Test failed: {e}")
except Exception as e:
print(f"\n✗ Error: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(main())
test_external_tools.py¶
"""Test external integration tools."""
import asyncio
from llm_helper import LLMHelper
from external_tools import ExternalTools
async def test_google_calendar():
"""Test Google Calendar integration."""
print("Testing Google Calendar...")
llm_helper = LLMHelper()
external_tools = ExternalTools(llm_helper)
try:
result = await external_tools.google_calendar_add(
summary="Test Event",
start_time="2025-10-01T10:00:00",
end_time="2025-10-01T11:00:00",
description="This is a test event"
)
if result["success"]:
print(f"✓ Calendar event created: {result}")
else:
print(f"Calendar test skipped or failed: {result['error']}")
except Exception as e:
print(f"Calendar test skipped (likely missing credentials): {e}")
async def test_github_pr():
"""Test GitHub PR creation."""
print("\nTesting GitHub PR...")
llm_helper = LLMHelper()
external_tools = ExternalTools(llm_helper)
try:
# Note: This will fail without a valid repo and token
result = await external_tools.github_create_pr(
repo_name="test/test-repo",
title="Test PR",
body="This is a test PR",
head_branch="test-branch",
base_branch="main"
)
if result["success"]:
print(f"✓ PR created: {result}")
else:
print(f"PR test expected to fail (test repo): {result['error']}")
except Exception as e:
print(f"PR test skipped (likely missing credentials): {e}")
async def test_datetime_parsing():
"""Test datetime parsing."""
print("\nTesting datetime parsing...")
llm_helper = LLMHelper()
external_tools = ExternalTools(llm_helper)
# Test invalid datetime
result = await external_tools.google_calendar_add(
summary="Test",
start_time="invalid-datetime",
end_time="2025-10-01T11:00:00"
)
assert not result["success"], "Should fail with invalid datetime"
print(f"✓ Invalid datetime handling works: {result['error']}")
# Test end before start
result = await external_tools.google_calendar_add(
summary="Test",
start_time="2025-10-01T11:00:00",
end_time="2025-10-01T10:00:00"
)
assert not result["success"], "Should fail when end is before start"
print(f"✓ Time validation works: {result['error']}")
async def main():
"""Run all tests."""
print("=== External Tools Tests ===\n")
try:
await test_datetime_parsing()
await test_google_calendar()
await test_github_pr()
print("\n✓ External tools tests completed!")
print("Note: Some tests may be skipped if credentials are not configured.")
except AssertionError as e:
print(f"\n✗ Test failed: {e}")
except Exception as e:
print(f"\n✗ Error: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(main())
test_file_tools.py¶
"""Test file system tools."""
import asyncio
import tempfile
from pathlib import Path
from llm_helper import LLMHelper
from file_tools import FileTools
from config import Config
async def test_file_write():
"""Test file write functionality."""
print("Testing file write...")
llm_helper = LLMHelper()
file_tools = FileTools(llm_helper)
# Test writing valid Python code
result = await file_tools.write_file(
path="test_output.py",
content='print("Hello, World!")\n',
overwrite=True
)
assert result["success"], f"File write failed: {result.get('error')}"
assert result["verification"] in ["passed", "skipped"]
print(f"✓ File write successful: {result}")
# Test syntax error detection
result = await file_tools.write_file(
path="test_syntax_error.py",
content='print("Unclosed string\n',
overwrite=True
)
if Config.AUTO_VERIFY_CODE:
assert not result["success"], "Should detect syntax error"
print(f"✓ Syntax error detected: {result['error']}")
else:
print("✓ Verification skipped (AUTO_VERIFY_CODE=False)")
async def test_file_edit():
"""Test file edit functionality."""
print("\nTesting file edit...")
llm_helper = LLMHelper()
file_tools = FileTools(llm_helper)
# Create a test file first
await file_tools.write_file(
path="test_edit.py",
content='message = "Hello"\nprint(message)\n',
overwrite=True
)
# Edit the file
result = await file_tools.edit_file(
path="test_edit.py",
search='message = "Hello"',
replace='message = "Hi there"'
)
assert result["success"], f"File edit failed: {result.get('error')}"
assert "diff_preview" in result
print(f"✓ File edit successful: {result}")
async def test_safety_checks():
"""Test safety checks for file operations."""
print("\nTesting safety checks...")
llm_helper = LLMHelper()
file_tools = FileTools(llm_helper)
# Test path outside workspace
result = await file_tools.write_file(
path="/tmp/outside_workspace.txt",
content="test"
)
# This should fail unless /tmp is in workspace
print(f"Path safety check result: {result}")
async def main():
"""Run all tests."""
print("=== File Tools Tests ===\n")
try:
await test_file_write()
await test_file_edit()
await test_safety_checks()
print("\n✓ All file tools tests passed!")
except AssertionError as e:
print(f"\n✗ Test failed: {e}")
except Exception as e:
print(f"\n✗ Error: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(main())
test_filesystem_enhanced.py¶
"""
Real tests for enhanced filesystem tools.
These tests perform actual file operations to verify functionality.
"""
import asyncio
import json
import pytest
import tempfile
import shutil
from pathlib import Path
import sys
# Add current directory to path
sys.path.insert(0, str(Path(__file__).parent))
# Mock Config for testing
class Config:
WORKSPACE_DIR = Path(tempfile.mkdtemp())
AUTO_VERIFY_CODE = False
REQUIRE_APPROVAL_FOR_DANGEROUS_OPS = False
# Set config before import
sys.modules['config'] = type(sys)('config')
sys.modules['config'].Config = Config
from filesystem_enhanced import FilesystemEnhanced
@pytest.fixture(scope="function")
def fs():
"""Create filesystem instance with temp workspace."""
# Create new temp dir for each test
Config.WORKSPACE_DIR = Path(tempfile.mkdtemp())
filesystem = FilesystemEnhanced()
yield filesystem
# Cleanup
if Config.WORKSPACE_DIR.exists():
try:
shutil.rmtree(Config.WORKSPACE_DIR)
except Exception:
pass
@pytest.fixture(scope="function")
def test_files(fs):
"""Create test files for testing."""
# Create some test files
try:
(Config.WORKSPACE_DIR / "test1.txt").write_text("Hello World")
(Config.WORKSPACE_DIR / "test2.txt").write_text("Python Testing\nLine 2\nLine 3")
(Config.WORKSPACE_DIR / "data.json").write_text('{"key": "value"}')
(Config.WORKSPACE_DIR / "subdir").mkdir()
(Config.WORKSPACE_DIR / "subdir" / "nested.txt").write_text("Nested file")
except Exception as e:
print(f"Error creating test files: {e}")
return fs
class TestReadOperations:
"""Tests for file reading operations."""
@pytest.mark.asyncio
async def test_read_text_file(self, test_files):
"""Test reading a text file."""
result = await test_files.read_text_file("test1.txt")
assert result["success"] is True
assert result["content"] == "Hello World"
assert result["file_size"] > 0
assert result["lines"] == 1
print("✅ Read text file successfully")
print(f" Content: {result['content']}")
@pytest.mark.asyncio
async def test_read_multiline_file(self, test_files):
"""Test reading multiline file."""
result = await test_files.read_text_file("test2.txt")
assert result["success"] is True
assert result["lines"] == 3
assert "Python Testing" in result["content"]
print("✅ Read multiline file")
print(f" Lines: {result['lines']}")
@pytest.mark.asyncio
async def test_read_nonexistent_file(self, fs):
"""Test reading non-existent file."""
result = await fs.read_text_file("nonexistent.txt")
assert result["success"] is False
assert "does not exist" in result["error"]
print("✅ Correctly handled nonexistent file")
@pytest.mark.asyncio
async def test_read_multiple_files(self, test_files):
"""Test reading multiple files at once."""
result = await test_files.read_multiple_files([
"test1.txt",
"test2.txt",
"data.json"
])
assert result["success"] is True
assert result["files_read"] == 3
assert result["files_failed"] == 0
assert "test1.txt" in result["results"]
assert "test2.txt" in result["results"]
print(f"✅ Read {result['files_read']} files")
print(f" Files: {', '.join(result['results'].keys())}")
@pytest.mark.asyncio
async def test_read_multiple_files_with_errors(self, test_files):
"""Test reading multiple files with some missing."""
result = await test_files.read_multiple_files([
"test1.txt",
"nonexistent.txt",
"test2.txt"
])
assert result["success"] is True # At least some succeeded
assert result["files_read"] == 2
assert result["files_failed"] == 1
assert len(result["errors"]) == 1
print(f"✅ Read {result['files_read']} files, {result['files_failed']} failed")
class TestListOperations:
"""Tests for directory listing operations."""
@pytest.mark.asyncio
async def test_list_directory_with_sizes(self, test_files):
"""Test listing directory with file sizes."""
result = await test_files.list_directory_with_sizes(".")
assert result["success"] is True
assert result["total_items"] >= 4 # At least 3 files + 1 dir
assert result["total_size"] > 0
assert len(result["contents"]) >= 4
# Check structure
item = result["contents"][0]
assert "name" in item
assert "type" in item
assert "size" in item
assert "size_human" in item
print(f"✅ Listed {result['total_items']} items")
print(f" Total size: {result['total_size_human']}")
@pytest.mark.asyncio
async def test_directory_tree(self, test_files):
"""Test generating directory tree."""
result = await test_files.directory_tree(".", max_depth=3)
assert result["success"] is True
assert "tree" in result
assert result["tree"]["type"] == "directory"
assert "children" in result["tree"]
assert len(result["tree"]["children"]) >= 4
print(f"✅ Generated directory tree")
print(f" Root: {result['root']}")
print(f" Items: {result['tree']['count']}")
@pytest.mark.asyncio
async def test_directory_tree_depth_limit(self, test_files):
"""Test directory tree with depth limit."""
# Create deeper structure
(Config.WORKSPACE_DIR / "deep" / "level2" / "level3").mkdir(parents=True)
result = await test_files.directory_tree(".", max_depth=2)
assert result["success"] is True
assert result["max_depth"] == 2
print("✅ Directory tree respects depth limit")
class TestSearchOperations:
"""Tests for file search operations."""
@pytest.mark.asyncio
async def test_search_files_pattern(self, test_files):
"""Test searching files by pattern."""
result = await test_files.search_files("*.txt", ".", recursive=True)
assert result["success"] is True
assert result["matches"] >= 3 # test1.txt, test2.txt, nested.txt
assert result["pattern"] == "*.txt"
# Check results structure
if len(result["files"]) > 0:
file_info = result["files"][0]
assert "path" in file_info
assert "size" in file_info
assert "size_human" in file_info
print(f"✅ Found {result['matches']} files matching *.txt")
@pytest.mark.asyncio
async def test_search_files_nonrecursive(self, test_files):
"""Test non-recursive search."""
result = await test_files.search_files("*.txt", ".", recursive=False)
assert result["success"] is True
assert result["recursive"] is False
# Should not find nested.txt
paths = [f["path"] for f in result["files"]]
assert not any("subdir" in p for p in paths)
print(f"✅ Non-recursive search: {result['matches']} files")
@pytest.mark.asyncio
async def test_search_files_json(self, test_files):
"""Test searching for specific file type."""
result = await test_files.search_files("*.json", ".", recursive=True)
assert result["success"] is True
assert result["matches"] >= 1
print(f"✅ Found {result['matches']} JSON files")
class TestFileInfo:
"""Tests for file information retrieval."""
@pytest.mark.asyncio
async def test_get_file_info(self, test_files):
"""Test getting file information."""
result = await test_files.get_file_info("test2.txt")
assert result["success"] is True
info = result["file_info"]
assert info["name"] == "test2.txt"
assert info["extension"] == ".txt"
assert info["is_file"] is True
assert info["is_directory"] is False
assert info["lines"] == 3
assert info["size"] > 0
print("✅ File info retrieved")
print(f" Name: {info['name']}")
print(f" Size: {info['size_human']}")
print(f" Lines: {info['lines']}")
@pytest.mark.asyncio
async def test_get_directory_info(self, test_files):
"""Test getting directory information."""
result = await test_files.get_file_info("subdir")
assert result["success"] is True
info = result["file_info"]
assert info["is_directory"] is True
assert info["is_file"] is False
print("✅ Directory info retrieved")
class TestMoveOperations:
"""Tests for move and copy operations."""
@pytest.mark.asyncio
async def test_move_file(self, test_files):
"""Test moving a file."""
result = await test_files.move_file("test1.txt", "moved.txt")
assert result["success"] is True
assert Path(Config.WORKSPACE_DIR / "moved.txt").exists()
assert not Path(Config.WORKSPACE_DIR / "test1.txt").exists()
print("✅ File moved successfully")
print(f" From: test1.txt")
print(f" To: moved.txt")
@pytest.mark.asyncio
async def test_move_file_no_overwrite(self, test_files):
"""Test move without overwrite."""
result = await test_files.move_file("test1.txt", "test2.txt", overwrite=False)
assert result["success"] is False
assert "already exists" in result["error"]
print("✅ Correctly prevented overwrite")
@pytest.mark.asyncio
async def test_copy_file(self, test_files):
"""Test copying a file."""
result = await test_files.copy_file("test1.txt", "copied.txt")
assert result["success"] is True
assert Path(Config.WORKSPACE_DIR / "copied.txt").exists()
assert Path(Config.WORKSPACE_DIR / "test1.txt").exists() # Original still exists
print("✅ File copied successfully")
@pytest.mark.asyncio
async def test_copy_directory(self, test_files):
"""Test copying a directory."""
result = await test_files.copy_file("subdir", "subdir_copy")
assert result["success"] is True
assert Path(Config.WORKSPACE_DIR / "subdir_copy").exists()
assert Path(Config.WORKSPACE_DIR / "subdir_copy" / "nested.txt").exists()
print("✅ Directory copied recursively")
class TestDeleteOperations:
"""Tests for delete operations."""
@pytest.mark.asyncio
async def test_delete_file(self, test_files):
"""Test deleting a file."""
result = await test_files.delete_file("test1.txt")
assert result["success"] is True
assert not Path(Config.WORKSPACE_DIR / "test1.txt").exists()
print("✅ File deleted")
@pytest.mark.asyncio
async def test_delete_directory_recursive(self, test_files):
"""Test deleting directory recursively."""
result = await test_files.delete_file("subdir", recursive=True)
assert result["success"] is True
assert not Path(Config.WORKSPACE_DIR / "subdir").exists()
print("✅ Directory deleted recursively")
@pytest.mark.asyncio
async def test_delete_directory_without_recursive(self, test_files):
"""Test that directory delete requires recursive flag."""
result = await test_files.delete_file("subdir", recursive=False)
assert result["success"] is False
assert "recursive" in result["error"].lower()
print("✅ Correctly required recursive flag")
class TestCreateOperations:
"""Tests for create operations."""
@pytest.mark.asyncio
async def test_create_directory(self, fs):
"""Test creating a directory."""
result = await fs.create_directory("newdir")
assert result["success"] is True
assert Path(Config.WORKSPACE_DIR / "newdir").exists()
print("✅ Directory created")
@pytest.mark.asyncio
async def test_create_nested_directory(self, fs):
"""Test creating nested directories."""
result = await fs.create_directory("parent/child/grandchild", parents=True)
assert result["success"] is True
assert Path(Config.WORKSPACE_DIR / "parent" / "child" / "grandchild").exists()
print("✅ Nested directories created")
@pytest.mark.asyncio
async def test_list_allowed_directories(self, fs):
"""Test listing allowed directories."""
result = await fs.list_allowed_directories()
assert result["success"] is True
assert result["count"] >= 1
assert len(result["allowed_directories"]) >= 1
print(f"✅ Listed {result['count']} allowed directories")
# Run tests
if __name__ == "__main__":
print("=" * 70)
print("Running Enhanced Filesystem Tools Tests")
print("=" * 70)
print()
pytest.main([__file__, "-v", "-s"])
test_multilang.py¶
"""Test multi-language code execution."""
import asyncio
import sys
from multilang_executor import LanguageExecutor, ExecutionStatus
async def test_language(executor: LanguageExecutor, language: str, code: str, description: str):
"""Test a specific language."""
print(f"\n{'='*60}")
print(f"Testing {language}: {description}")
print(f"{'='*60}")
result = await executor.execute_code(code, language, timeout=10.0)
print(f"Status: {result.get('status')}")
if result.get('stdout'):
print(f"Output:\n{result['stdout']}")
if result.get('stderr'):
print(f"Errors:\n{result['stderr']}")
if result.get('compile_output'):
print(f"Compile output:\n{result['compile_output']}")
success = result.get('status') == ExecutionStatus.SUCCESS
print(f"✅ PASSED" if success else f"❌ FAILED")
return success
async def main():
"""Run all language tests."""
executor = LanguageExecutor()
tests = [
# Python
("python", """
import numpy as np
import pandas as pd
data = pd.DataFrame({'A': [1, 2, 3], 'B': [4, 5, 6]})
print("Data shape:", data.shape)
print("Mean of A:", data['A'].mean())
""", "NumPy and Pandas"),
# JavaScript
("javascript", """
console.log('Hello from Node.js!');
const numbers = [1, 2, 3, 4, 5];
const sum = numbers.reduce((a, b) => a + b, 0);
console.log('Sum:', sum);
""", "Array operations"),
# TypeScript
("typescript", """
interface Point {
x: number;
y: number;
}
const point: Point = { x: 10, y: 20 };
console.log(`Point: (${point.x}, ${point.y})`);
""", "Type-safe interfaces"),
# Go
("go", """
package main
import "fmt"
func main() {
fmt.Println("Hello from Go!")
sum := 0
for i := 1; i <= 10; i++ {
sum += i
}
fmt.Printf("Sum of 1-10: %d\\n", sum)
}
""", "Loops and formatting"),
# Java
("java", """
public class Main {
public static void main(String[] args) {
System.out.println("Hello from Java!");
int sum = 0;
for (int i = 1; i <= 10; i++) {
sum += i;
}
System.out.println("Sum of 1-10: " + sum);
}
}
""", "Basic class and loops"),
# C++
("cpp", """
#include <iostream>
#include <vector>
#include <numeric>
int main() {
std::cout << "Hello from C++!" << std::endl;
std::vector<int> numbers = {1, 2, 3, 4, 5};
int sum = std::accumulate(numbers.begin(), numbers.end(), 0);
std::cout << "Sum: " << sum << std::endl;
return 0;
}
""", "STL vector and accumulate"),
# Rust
("rust", """
fn main() {
println!("Hello from Rust!");
let numbers = vec![1, 2, 3, 4, 5];
let sum: i32 = numbers.iter().sum();
println!("Sum: {}", sum);
}
""", "Vector and iterators"),
# PHP
("php", """
<?php
echo "Hello from PHP!\\n";
$numbers = [1, 2, 3, 4, 5];
$sum = array_sum($numbers);
echo "Sum: $sum\\n";
?>
""", "Array operations"),
# Bash
("bash", """
echo "Hello from Bash!"
sum=0
for i in {1..10}; do
sum=$((sum + i))
done
echo "Sum of 1-10: $sum"
""", "Shell loops"),
]
print(f"\n{'#'*60}")
print(f"# Multi-Language Code Execution Test Suite")
print(f"{'#'*60}")
results = []
for language, code, description in tests:
try:
success = await test_language(executor, language, code, description)
results.append((language, success))
except Exception as e:
print(f"❌ EXCEPTION: {e}")
results.append((language, False))
# Summary
print(f"\n{'='*60}")
print(f"Test Summary")
print(f"{'='*60}")
passed = sum(1 for _, success in results if success)
total = len(results)
for language, success in results:
status = "✅ PASS" if success else "❌ FAIL"
print(f"{status} - {language}")
print(f"\nTotal: {passed}/{total} passed ({100*passed//total}%)")
return passed == total
if __name__ == "__main__":
success = asyncio.run(main())
sys.exit(0 if success else 1)
test_multilang_output.py¶
import asyncio
import pytest
from multilang_executor import get_all_output, LanguageExecutor, ExecutionStatus
@pytest.mark.asyncio
async def test_get_all_output_reads_full_payload():
class FakeStream:
def __init__(self, data: bytes):
self._data = data
self._done = False
async def read(self, n: int = -1):
if self._done:
return b""
self._done = True
return self._data
payload = b"hello world" * 1000
out = await get_all_output(FakeStream(payload))
assert out == payload.decode()
@pytest.mark.asyncio
async def test_execute_captures_stdout():
exe = LanguageExecutor()
result = await exe.execute_code("print('ok-from-executor')", "python", timeout=10)
assert result["status"] == ExecutionStatus.SUCCESS
assert "ok-from-executor" in result["stdout"]
test_null_language.py¶
"""Null optional language must default to python on public paths."""
import asyncio
import tempfile
from pathlib import Path
from unittest.mock import MagicMock
from multilang_executor import LanguageExecutor
from execution_tools import ExecutionTools
def test_null_language_executor_defaults_to_python():
le = LanguageExecutor(workspace_dir=Path(tempfile.mkdtemp()))
result = asyncio.run(le.execute_code("print(42)", language=None, timeout=10))
assert isinstance(result, dict)
assert result.get("language") == "python"
assert "42" in (result.get("stdout") or "")
def test_null_language_code_interpreter_defaults(monkeypatch):
"""Public MCP path: ExecutionTools.code_interpreter(language=None)."""
helper = MagicMock()
et = ExecutionTools(helper)
seen = {}
async def fake_exec(code, language, timeout=30.0, stdin=None, files=None):
seen["language"] = language
return {
"success": True,
"stdout": "ok\n",
"stderr": "",
"language": language,
"returncode": 0,
"status": "ok",
}
monkeypatch.setattr(et.lang_executor, "execute_code", fake_exec)
import config as cfg
monkeypatch.setattr(cfg.Config, "AUTO_VERIFY_CODE", False, raising=False)
monkeypatch.setattr(cfg.Config, "REQUIRE_APPROVAL_FOR_DANGEROUS_OPS", False, raising=False)
monkeypatch.setattr(cfg.Config, "AUTO_SUMMARIZE_OUTPUT", False, raising=False)
out = asyncio.run(et.code_interpreter("print(1)", language=None))
assert seen["language"] == "python"
assert out["language"] == "python"
assert out.get("error") in (None, "")
test_script.py¶
test_terminal_controller.py¶
"""
Tests for Terminal Controller.
Tests command execution with directory navigation and file operations.
"""
import asyncio
import json
import pytest
import tempfile
import shutil
from pathlib import Path
import sys
# Mock Config for testing
class Config:
WORKSPACE_DIR = Path(tempfile.mkdtemp())
AUTO_VERIFY_CODE = False
REQUIRE_APPROVAL_FOR_DANGEROUS_OPS = False
sys.modules['config'] = type(sys)('config')
sys.modules['config'].Config = Config
from terminal_controller import TerminalController
@pytest.fixture(scope="function")
def tc():
"""Create terminal controller with temp workspace."""
Config.WORKSPACE_DIR = Path(tempfile.mkdtemp())
controller = TerminalController()
yield controller
# Cleanup
if Config.WORKSPACE_DIR.exists():
try:
shutil.rmtree(Config.WORKSPACE_DIR)
except Exception:
pass
class TestTerminalBasics:
"""Tests for basic terminal operations."""
@pytest.mark.asyncio
async def test_get_current_directory(self, tc):
"""Test getting current directory."""
result = await tc.get_current_directory()
assert result["success"] is True
assert "current_directory" in result
assert "workspace" in result
print(f"✅ Current directory: {result['current_directory']}")
@pytest.mark.asyncio
async def test_execute_command_simple(self, tc):
"""Test executing a simple command."""
result = await tc.execute_command("echo 'Hello World'")
assert result["success"] is True
assert "Hello World" in result["stdout"]
assert result["returncode"] == 0
print(f"✅ Command executed: {result['command']}")
@pytest.mark.asyncio
async def test_execute_command_ls(self, tc):
"""Test listing directory."""
result = await tc.execute_command("ls")
assert result["success"] is True
print(f"✅ Directory listing completed")
@pytest.mark.asyncio
async def test_command_history(self, tc):
"""Test command history."""
await tc.execute_command("echo 'test1'")
await tc.execute_command("echo 'test2'")
await tc.execute_command("echo 'test3'")
result = await tc.get_command_history(count=2)
assert result["success"] is True
assert len(result["history"]) == 2
assert result["total"] == 3
print(f"✅ Command history: {result['count']} recent commands")
class TestDirectoryOperations:
"""Tests for directory navigation."""
@pytest.mark.asyncio
async def test_change_directory(self, tc):
"""Test changing directory."""
# Create a subdirectory
subdir = Config.WORKSPACE_DIR / "subdir"
subdir.mkdir()
result = await tc.change_directory("subdir")
assert result["success"] is True
assert "subdir" in result["current_directory"]
print(f"✅ Changed to: {result['current_directory']}")
@pytest.mark.asyncio
async def test_list_directory(self, tc):
"""Test listing directory."""
# Create some files
(Config.WORKSPACE_DIR / "file1.txt").write_text("test")
(Config.WORKSPACE_DIR / "file2.txt").write_text("test")
result = await tc.list_directory(".")
assert result["success"] is True
assert result["count"] >= 2
print(f"✅ Listed {result['count']} items")
@pytest.mark.asyncio
async def test_change_to_nonexistent(self, tc):
"""Test changing to nonexistent directory."""
result = await tc.change_directory("nonexistent")
assert result["success"] is False
assert "does not exist" in result["error"]
print("✅ Correctly rejected nonexistent directory")
class TestFileOperations:
"""Tests for file operations through terminal controller."""
@pytest.mark.asyncio
async def test_write_file(self, tc):
"""Test writing a file."""
result = await tc.write_file("test.txt", "Hello Terminal")
assert result["success"] is True
assert result["bytes_written"] > 0
# Verify file exists
file_path = Config.WORKSPACE_DIR / "test.txt"
assert file_path.exists()
assert file_path.read_text() == "Hello Terminal"
print(f"✅ Wrote {result['bytes_written']} bytes")
@pytest.mark.asyncio
async def test_read_file(self, tc):
"""Test reading a file."""
# Create a file
test_file = Config.WORKSPACE_DIR / "read_test.txt"
test_file.write_text("Test content\nLine 2")
result = await tc.read_file("read_test.txt")
assert result["success"] is True
assert result["content"] == "Test content\nLine 2"
assert result["lines"] == 2
print(f"✅ Read file: {result['lines']} lines")
@pytest.mark.asyncio
async def test_insert_file_content(self, tc):
"""Test inserting content into file."""
# Create a file
test_file = Config.WORKSPACE_DIR / "insert_test.txt"
test_file.write_text("Line 1\nLine 3")
result = await tc.insert_file_content("insert_test.txt", "Line 2", 2)
assert result["success"] is True
assert result["line_number"] == 2
# Verify content
content = test_file.read_text()
lines = content.splitlines()
assert lines[1] == "Line 2"
print("✅ Inserted content at line 2")
@pytest.mark.asyncio
async def test_delete_file_content(self, tc):
"""Test deleting lines from file."""
# Create a file
test_file = Config.WORKSPACE_DIR / "delete_test.txt"
test_file.write_text("Line 1\nLine 2\nLine 3\nLine 4")
result = await tc.delete_file_content("delete_test.txt", 2, 3)
assert result["success"] is True
assert result["deleted_lines"] == 2
# Verify content
content = test_file.read_text()
assert "Line 2" not in content
assert "Line 3" not in content
assert "Line 1" in content
assert "Line 4" in content
print(f"✅ Deleted {result['deleted_lines']} lines")
@pytest.mark.asyncio
async def test_update_file_content(self, tc):
"""Test updating a line in file."""
# Create a file
test_file = Config.WORKSPACE_DIR / "update_test.txt"
test_file.write_text("Line 1\nOld Line 2\nLine 3")
result = await tc.update_file_content("update_test.txt", 2, "New Line 2")
assert result["success"] is True
assert result["old_content"] == "Old Line 2"
assert result["new_content"] == "New Line 2"
# Verify content
content = test_file.read_text()
assert "New Line 2" in content
assert "Old Line 2" not in content
print("✅ Updated line 2")
if __name__ == "__main__":
print("=" * 70)
print("Running Terminal Controller Tests")
print("=" * 70)
print()
pytest.main([__file__, "-v", "-s"])