127 lines
4.0 KiB
Python
127 lines
4.0 KiB
Python
"""
|
||
终端命令执行 MCP Server
|
||
通过 MCP 协议暴露 run_command 命令，供 Dev Agent 使用
|
||
"""
|
||
import asyncio
|
||
import subprocess
|
||
import shlex
|
||
import os
|
||
import sys
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from mcp.server import Server
|
||
from mcp.server.stdio import stdio_server
|
||
from mcp.types import Tool, TextContent
|
||
|
||
|
||
def log(msg):
|
||
"""写日志到 stderr，不污染 MCP stdio 协议通道"""
|
||
print(msg, file=sys.stderr, flush=True)
|
||
|
||
|
||
app = Server("command-executor")
|
||
|
||
|
||
@app.list_tools()
|
||
async def list_tools():
|
||
return [
|
||
Tool(
|
||
name="run_command",
|
||
description="在终端中执行一条命令，返回输出结果。适用于编译、构建、运行测试等场景。",
|
||
inputSchema={
|
||
"type": "object",
|
||
"properties": {
|
||
"command": {
|
||
"type": "string",
|
||
"description": "要执行的命令，如 'npm run build'、'npm test' 等",
|
||
},
|
||
"cwd": {
|
||
"type": "string",
|
||
"description": "命令执行的工作目录，默认使用当前目录",
|
||
},
|
||
"timeout": {
|
||
"type": "integer",
|
||
"description": "超时时间（秒），默认 300 秒",
|
||
"default": 300,
|
||
},
|
||
},
|
||
"required": ["command"],
|
||
},
|
||
),
|
||
]
|
||
|
||
|
||
@app.call_tool()
|
||
async def call_tool(name: str, arguments: dict):
|
||
if name != "run_command":
|
||
raise ValueError(f"Unknown tool: {name}")
|
||
|
||
cmd = arguments.get("command", "")
|
||
cwd = arguments.get("cwd") or os.getcwd()
|
||
timeout = int(arguments.get("timeout", 300))
|
||
|
||
if not cmd:
|
||
return [TextContent(type="text", text="错误：命令不能为空")]
|
||
|
||
log(f" [command] {cmd}")
|
||
log(f" [cwd] {cwd}")
|
||
log(f" [timeout] {timeout}s")
|
||
log(f" [PATH] {os.environ.get('PATH', 'N/A')[:200]}")
|
||
log(f" [where npm] {__import__('shutil').which('npm.cmd')}")
|
||
|
||
try:
|
||
# 使用异步 subprocess，避免阻塞 asyncio 事件循环
|
||
# 在 Windows 上，npm 是 .cmd 文件，需要通过 shell 执行
|
||
proc = await asyncio.create_subprocess_shell(
|
||
cmd,
|
||
cwd=cwd,
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
)
|
||
stdout_bytes, stderr_bytes = await asyncio.wait_for(
|
||
proc.communicate(),
|
||
timeout=timeout,
|
||
)
|
||
|
||
stdout = stdout_bytes.decode("utf-8", errors="replace")
|
||
stderr = stderr_bytes.decode("utf-8", errors="replace")
|
||
|
||
output_parts = []
|
||
if stdout:
|
||
output_parts.append(f"[stdout]\n{stdout}")
|
||
if stderr:
|
||
output_parts.append(f"[stderr]\n{stderr}")
|
||
|
||
output = "\n".join(output_parts) if output_parts else "(无输出)"
|
||
|
||
max_len = 10000
|
||
if len(output) > max_len:
|
||
output = output[:max_len] + f"\n\n...（输出已截断，共 {len(output)} 字符）"
|
||
|
||
status = "成功" if proc.returncode == 0 else f"失败 (退出码 {proc.returncode})"
|
||
log(f" [result] {status}")
|
||
return [TextContent(type="text", text=f"命令执行{status}\n{output}")]
|
||
|
||
except asyncio.TimeoutError:
|
||
# 超时后杀进程
|
||
proc.kill()
|
||
await proc.wait()
|
||
log(f" [error] 超时")
|
||
return [TextContent(type="text", text=f"命令执行超时（{timeout}秒）: {cmd}")]
|
||
except Exception as e:
|
||
log(f" [error] {e}")
|
||
return [TextContent(type="text", text=f"命令执行出错: {e}")]
|
||
|
||
|
||
async def main():
|
||
log("MCP 命令执行服务器启动中...")
|
||
async with stdio_server() as (read_stream, write_stream):
|
||
await app.run(
|
||
read_stream, write_stream,
|
||
app.create_initialization_options()
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main()) |