""" Dev Agent MCP Server（stdio 模式） 通过标准输入输出与 MCP 客户端通信，适合 CodeBuddy 本地使用。 用法：python -m mcp_server.stdio_server """ import os import sys import asyncio from dotenv import load_dotenv # 确保项目根目录在 path 里 PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) if PROJECT_ROOT not in sys.path: sys.path.insert(0, PROJECT_ROOT) from mcp.server import Server from mcp.types import Tool, TextContent from mcp.server.stdio import stdio_server from .task_manager import TaskManager from .agent_runner import AgentRunner # 加载 .env load_dotenv(os.path.join(PROJECT_ROOT, "my_agent", ".env")) # 配置 DATA_DIR = os.getenv("MCP_DATA_DIR", os.path.join(PROJECT_ROOT, "data")) # 初始化组件 task_manager = TaskManager(store_dir=os.path.join(DATA_DIR, "tasks")) agent_runner = AgentRunner() mcp_server = Server("dev-agent-mcp-server") # --- MCP 工具定义 --- @mcp_server.list_tools() async def list_tools(): return [ Tool( name="submit_task", description=( "提交一个开发任务给 Dev Agent 执行。任务将异步执行，" "提交后返回 task_id，用 get_task_status 查询进度。" ), inputSchema={ "type": "object", "properties": { "description": { "type": "string", "description": "任务的详细描述，要做什么开发工作", }, "project_path": { "type": "string", "description": "项目的本地路径，agent 将在此目录下工作", }, "requirements": { "type": "string", "description": "（可选）额外的要求或约束条件", }, }, "required": ["description", "project_path"], }, ), Tool( name="get_task_status", description="查询任务的当前状态（pending/running/completed/failed/cancelled）", inputSchema={ "type": "object", "properties": { "task_id": { "type": "string", "description": "任务 ID", }, }, "required": ["task_id"], }, ), Tool( name="get_task_result", description="获取任务的执行结果（完成后调用）", inputSchema={ "type": "object", "properties": { "task_id": { "type": "string", "description": "任务 ID", }, }, "required": ["task_id"], }, ), Tool( name="get_task_log", description="获取任务的执行日志", inputSchema={ "type": "object", "properties": { "task_id": { "type": "string", "description": "任务 ID", }, }, "required": ["task_id"], }, ), Tool( name="cancel_task", description="取消一个正在执行或等待中的任务", inputSchema={ "type": "object", "properties": { "task_id": { "type": "string", "description": "任务 ID", }, }, "required": ["task_id"], }, ), Tool( name="list_tasks", description="列出所有任务，可按状态过滤", inputSchema={ "type": "object", "properties": { "status": { "type": "string", "description": "（可选）按状态过滤：pending/running/completed/failed/cancelled", }, "limit": { "type": "integer", "description": "（可选）返回数量限制，默认 20", "default": 20, }, }, }, ), ] # --- MCP 工具实现 --- @mcp_server.call_tool() async def call_tool(name: str, arguments: dict): if name == "submit_task": return await _submit_task(arguments) elif name == "get_task_status": return _get_task_status(arguments) elif name == "get_task_result": return _get_task_result(arguments) elif name == "get_task_log": return _get_task_log(arguments) elif name == "cancel_task": return await _cancel_task(arguments) elif name == "list_tasks": return _list_tasks(arguments) else: return [TextContent(type="text", text=f"错误：未知工具 {name}")] async def _submit_task(args: dict): description = args.get("description", "") project_path = args.get("project_path", "") requirements = args.get("requirements", "") if not description: return [TextContent(type="text", text="错误：description 不能为空")] if not project_path: return [TextContent(type="text", text="错误：project_path 不能为空")] if not os.path.isdir(project_path): return [TextContent(type="text", text=f"错误：项目路径不存在 {project_path}")] task = await task_manager.submit_task( description=description, project_path=project_path, requirements=requirements, ) return [TextContent( type="text", text=( f"任务已提交\n" f"任务ID: {task['id']}\n" f"状态: {task['status']}\n" f"描述: {description[:100]}\n" f"项目: {project_path}\n" f"\n" f"请使用 get_task_status 查询进度。" ), )] def _get_task_status(args: dict): task_id = args.get("task_id", "") task = task_manager.get_task(task_id) if not task: return [TextContent(type="text", text=f"错误：任务不存在 {task_id}")] return [TextContent( type="text", text=( f"任务状态\n" f"任务ID: {task['id']}\n" f"状态: {task['status']}\n" f"描述: {task['description'][:100]}\n" f"创建时间: {_format_time(task.get('created_at'))}\n" f"更新时间: {_format_time(task.get('updated_at'))}\n" ), )] def _get_task_result(args: dict): task_id = args.get("task_id", "") task = task_manager.get_task(task_id) if not task: return [TextContent(type="text", text=f"错误：任务不存在 {task_id}")] result = task.get("result") status = task["status"] if status in ("pending", "running"): return [TextContent( type="text", text=( f"任务尚未完成（状态：{status}），" f"请稍后再试或使用 get_task_status 查询进度。" ), )] if not result: return [TextContent(type="text", text=f"任务结果为空，状态：{status}")] if isinstance(result, dict): summary = result.get("summary", str(result)) tool_count = result.get("tool_calls_count", 0) full = result.get("full_response", "") return [TextContent( type="text", text=( f"任务结果（{status}）\n" f"{'='*40}\n" f"{summary}\n" f"{'='*40}\n" f"工具调用次数: {tool_count}\n" f"\n" f"--- 完整回复 ---\n" f"{full[:5000]}" f"\n{'...' if len(full) > 5000 else ''}" ), )] return [TextContent(type="text", text=str(result))] def _get_task_log(args: dict): task_id = args.get("task_id", "") task = task_manager.get_task(task_id) if not task: return [TextContent(type="text", text=f"错误：任务不存在 {task_id}")] logs = task.get("logs", []) if not logs: return [TextContent(type="text", text="暂无日志")] lines = [] for log in logs[-50:]: lines.append(f"[{log['time']}] {log['message']}") return [TextContent(type="text", text="\n".join(lines))] async def _cancel_task(args: dict): task_id = args.get("task_id", "") success = await task_manager.cancel_task(task_id) if success: return [TextContent(type="text", text=f"任务 {task_id} 已取消")] else: return [TextContent(type="text", text=f"取消失败：任务不存在或已结束")] def _list_tasks(args: dict): status = args.get("status") limit = int(args.get("limit", 20)) tasks = task_manager.list_tasks(status=status) tasks = tasks[:limit] if not tasks: return [TextContent(type="text", text="没有找到任务")] lines = [f"任务列表（共 {len(tasks)} 个）:"] for t in tasks: lines.append( f" [{t['status']}] {t['id']} - {t['description'][:50]} " f"({_format_time(t.get('created_at'))})" ) return [TextContent(type="text", text="\n".join(lines))] def _format_time(ts: float = None) -> str: import time if not ts: return "-" return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts)) async def _task_executor(task: dict) -> dict: """任务执行器，交给 TaskManager 调用""" return await agent_runner.run_task(task) async def main(): """启动 stdio MCP Server""" # 日志写 stderr，不污染 stdout（MCP 协议通道） print("Dev Agent MCP Server (stdio) 启动中...", file=sys.stderr) # 启动任务管理器 await task_manager.start(executor=_task_executor) try: async with stdio_server() as (read_stream, write_stream): await mcp_server.run( read_stream, write_stream, mcp_server.create_initialization_options(), ) finally: await task_manager.stop() print("Dev Agent MCP Server 已停止", file=sys.stderr) if __name__ == "__main__": asyncio.run(main())