From d3e0861a4058a0885bac13a42604e371519a7f9e Mon Sep 17 00:00:00 2001 From: handsomeAq Date: Wed, 5 Aug 2026 16:32:44 +0800 Subject: [PATCH] =?UTF-8?q?agent1.4=E7=89=88=E6=9C=AC=20=E7=B2=BE=E7=AE=80?= =?UTF-8?q?=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- agents/luna/api_server.py | 4 + agents/my_agent/api_server.py | 4 +- agents/my_agent/chat.py | 2 +- agents/my_agent/task_receiver.py | 120 -------- agents/qwen/api_server.py | 4 + gateway_client.py | 81 ++++- mcp_server/__init__.py | 1 - mcp_server/agent_runner.py | 153 ---------- mcp_server/server.py | 408 ------------------------- mcp_server/start.py | 25 -- mcp_server/stdio_server.py | 337 -------------------- mcp_server/task_manager.py | 164 ---------- mcp_server/task_store.py | 56 ---- mcp_tools/__init__.py | 1 - mcp_tools/command_executor/__init__.py | 1 - mcp_tools/command_executor/server.py | 127 -------- 16 files changed, 90 insertions(+), 1398 deletions(-) delete mode 100644 agents/my_agent/task_receiver.py delete mode 100644 mcp_server/__init__.py delete mode 100644 mcp_server/agent_runner.py delete mode 100644 mcp_server/server.py delete mode 100644 mcp_server/start.py delete mode 100644 mcp_server/stdio_server.py delete mode 100644 mcp_server/task_manager.py delete mode 100644 mcp_server/task_store.py delete mode 100644 mcp_tools/__init__.py delete mode 100644 mcp_tools/command_executor/__init__.py delete mode 100644 mcp_tools/command_executor/server.py diff --git a/agents/luna/api_server.py b/agents/luna/api_server.py index 7710fb8..332d364 100644 --- a/agents/luna/api_server.py +++ b/agents/luna/api_server.py @@ -110,6 +110,10 @@ def main(): api_server = create_api_server() fastapi_app = api_server.get_fast_api_app() + # 挂载 A2A 网关任务接收端点（POST /tasks/{request_id}） + from task_receiver import create_task_router + fastapi_app.include_router(create_task_router(dev_app)) + # 向 A2A 网关注册并启动心跳（注册失败不阻塞服务启动） gateway_ok = gateway_client.register_agent(dev_app.name, f"http://127.0.0.1:{PORT}") if gateway_ok: diff --git a/agents/my_agent/api_server.py b/agents/my_agent/api_server.py index 8add327..b57bb39 100644 --- a/agents/my_agent/api_server.py +++ b/agents/my_agent/api_server.py @@ -111,8 +111,8 @@ def main(): fastapi_app = api_server.get_fast_api_app() # 挂载 A2A 网关任务接收端点（POST /tasks/{request_id}） - from agents.my_agent.task_receiver import router as task_router - fastapi_app.include_router(task_router) + from task_receiver import create_task_router + fastapi_app.include_router(create_task_router(dev_app)) # 向 A2A 网关注册并启动心跳（注册失败不阻塞服务启动） gateway_ok = gateway_client.register_agent(dev_app.name, f"http://127.0.0.1:{PORT}") diff --git a/agents/my_agent/chat.py b/agents/my_agent/chat.py index b58b514..8549a60 100644 --- a/agents/my_agent/chat.py +++ b/agents/my_agent/chat.py @@ -131,7 +131,7 @@ async def chat(session_id: str | None = None): print("再见！") break - print("花花: ", end="", flush=True) + print("agent: ", end="", flush=True) async def _agent_task(): """运行 agent 并流式输出，返回是否完成""" diff --git a/agents/my_agent/task_receiver.py b/agents/my_agent/task_receiver.py deleted file mode 100644 index fee0742..0000000 --- a/agents/my_agent/task_receiver.py +++ /dev/null @@ -1,120 +0,0 @@ -"""A2A 网关任务接收端点：接收网关主动推送的任务，后台执行 agent，完成后回传结果。 - -契约（网关 relay.py dispatch_command 推送）: - POST {endpoint}/tasks/{request_id} - body: {"auth": GATEWAY_AUTH, "request_id": str, "payload": {...}} - 成功响应 202（立即确认），执行完成后由后台线程回传网关 /api/agent/result。 -""" -import asyncio -import logging -import threading - -from fastapi import APIRouter, HTTPException, Request -from fastapi.responses import JSONResponse - -import gateway_client -from agents.my_agent.app import dev_app -from google.adk.runners import InMemoryRunner -from google.genai import types as genai_types - -logger = logging.getLogger(__name__) - -router = APIRouter(tags=["tasks"]) - -# 与 api_server.py 一致的导入路径（独立运行时兜底） -import os -import sys - -PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) -if PROJECT_ROOT not in sys.path: - sys.path.insert(0, PROJECT_ROOT) - - -def _payload_to_prompt(payload: dict) -> str: - """将网关任务 payload 转换为 agent 的用户指令。""" - if not payload: - return "请执行任务并汇报结果。" - if "prompt" in payload and payload["prompt"]: - return str(payload["prompt"]) - if "cmd" in payload and payload["cmd"]: - return f"请执行以下命令并汇报执行结果：\n{payload['cmd']}" - # 兜底：序列化整个 payload - return "请根据以下任务载荷执行并汇报结果：\n" + str(payload) - - -async def _run_agent_once(prompt: str, request_id: str) -> tuple[str, str]: - """用 ADK InMemoryRunner 运行一次 agent，返回 (接受任务后的首条回复, 最终总结)。 - - dev_app 是 App 容器（根 agent 非裸 LlmAgent），run_async 不会自动创建 - session，需先用 runner.session_service 显式创建。 - """ - runner = InMemoryRunner(app=dev_app) - session_id = f"task-{request_id}" - await runner.session_service.create_session( - app_name=runner.app_name, - user_id="gateway", - session_id=session_id, - ) - texts: list[str] = [] - final_text = "" - async for event in runner.run_async( - user_id="gateway", - session_id=session_id, - new_message=genai_types.Content(role="user", parts=[genai_types.Part(text=prompt)]), - ): - if event.is_final_response(): - if event.content and event.content.parts: - for part in event.content.parts: - text = getattr(part, "text", None) - if text: - final_text += text - break - if event.content and event.content.parts: - for part in event.content.parts: - text = getattr(part, "text", None) - if text: - texts.append(text) - # 首条回复 = 接受任务后的第一条回应；最终总结 = final response - reply = (texts[0] if texts else final_text).strip() - summary = final_text.strip() or "\n".join(t for t in texts if t).strip() or "(无输出)" - return reply, summary - - -def _execute_and_report(request_id: str, payload: dict) -> None: - """后台线程：执行 agent，成功后回传 success，异常回传 failed。""" - try: - prompt = _payload_to_prompt(payload) - reply, summary = asyncio.run(_run_agent_once(prompt, request_id)) - gateway_client.report_result( - request_id, - agent_id=dev_app.name, - status="success", - progress=100, - result={"reply": reply, "output": summary}, - ) - except Exception as e: - logger.exception("agent task failed request=%s", request_id) - gateway_client.report_result( - request_id, - agent_id=dev_app.name, - status="failed", - progress=100, - error_info=str(e), - ) - - -@router.post("/tasks/{request_id}") -async def receive_task(request_id: str, request: Request): - """接收网关推送的任务，立即 202 确认，后台执行。""" - body = await request.json() - if body.get("auth") != gateway_client.GATEWAY_AUTH: - raise HTTPException(status_code=401, detail="invalid auth") - payload = body.get("payload") or {} - threading.Thread( - target=_execute_and_report, - args=(request_id, payload), - name=f"task-{request_id[:8]}", - daemon=True, - ).start() - logger.info("task received request=%s payload=%s", request_id, payload) - return JSONResponse(status_code=202, content={"ok": True, "request_id": request_id, "status": "accepted"}) diff --git a/agents/qwen/api_server.py b/agents/qwen/api_server.py index 5ab290e..748d9ef 100644 --- a/agents/qwen/api_server.py +++ b/agents/qwen/api_server.py @@ -110,6 +110,10 @@ def main(): api_server = create_api_server() fastapi_app = api_server.get_fast_api_app() + # 挂载 A2A 网关任务接收端点（POST /tasks/{request_id}） + from task_receiver import create_task_router + fastapi_app.include_router(create_task_router(dev_app)) + # 向 A2A 网关注册并启动心跳（注册失败不阻塞服务启动） gateway_ok = gateway_client.register_agent(dev_app.name, f"http://127.0.0.1:{PORT}") if gateway_ok: diff --git a/gateway_client.py b/gateway_client.py index f7f5c18..d7c8337 100644 --- a/gateway_client.py +++ b/gateway_client.py @@ -28,6 +28,82 @@ HEARTBEAT_INTERVAL = int(os.getenv("HEARTBEAT_INTERVAL", "10")) _client = httpx.Client(timeout=5.0) +# 取消/停止指令下发：网关复用 CLI 的 /api/cli/events 通道（event=task_stop） +# Agent 通过同一个 cli_session_id 订阅，收到匹配自身正在执行 request_id 的 task_stop 时置停止标志。 +_stop_registry: dict[str, threading.Event] = {} + + +def mark_stop_requested(request_id: str) -> None: + """标记某任务需要停止（收到 task_stop 后调用）。""" + ev = _stop_registry.get(request_id) + if ev is None: + ev = threading.Event() + _stop_registry[request_id] = ev + ev.set() + + +def clear_stop_requested(request_id: str) -> None: + """任务开始执行前清除停止标志。""" + ev = _stop_registry.get(request_id) + if ev is not None: + ev.clear() + + +def is_stop_requested(request_id: str) -> bool: + """判断某任务是否已被要求停止（供执行循环轮询检查）。""" + ev = _stop_registry.get(request_id) + return ev is not None and ev.is_set() + + +def wait_stop(request_id: str, timeout: float = 0.5) -> bool: + """等待停止标志，返回 True 表示已收到停止请求。执行循环可用它做可中断 sleep。""" + ev = _stop_registry.get(request_id) + if ev is None: + ev = threading.Event() + _stop_registry[request_id] = ev + return ev.wait(timeout) + + +def _sse_subscribe_poll(cli_session_id: str) -> None: + """后台线程：订阅网关 /api/cli/events 通道，识别 task_stop 指令并标记停止。""" + url = f"{GATEWAY_URL}/api/cli/events?cli_session_id={cli_session_id}&auth={GATEWAY_AUTH}" + while True: + try: + with _client.stream("GET", url, timeout=None) as resp: + if resp.status_code != 200: + logger.warning("sse subscribe failed status=%s", resp.status_code) + time.sleep(HEARTBEAT_INTERVAL) + continue + for line in resp.iter_lines(): + if not line or not line.startswith("data:"): + continue + try: + import json + evt = json.loads(line[len("data:"):].strip()) + except Exception: + continue + if evt.get("event") == "task_stop": + rid = evt.get("request_id") + if rid: + mark_stop_requested(rid) + logger.info("stop requested received request=%s", rid) + except Exception as e: + logger.warning("sse subscribe loop error err=%s", e) + time.sleep(HEARTBEAT_INTERVAL) + + +def start_stop_listener(cli_session_id: str) -> threading.Thread: + """启动 SSE 停止指令订阅线程（daemon）。""" + t = threading.Thread( + target=_sse_subscribe_poll, + args=(cli_session_id,), + name=f"gateway-sse-{cli_session_id[:8]}", + daemon=True, + ) + t.start() + logger.info("stop listener started session=%s", cli_session_id) + return t + def _parse_tags() -> list[str]: raw = os.getenv("AGENT_TAGS", "code") @@ -62,7 +138,7 @@ def _heartbeat_once(agent_id: str, current_load: int) -> bool: try: resp = _client.post( f"{GATEWAY_URL}/api/agent/heartbeat", - json={"agent_id": agent_id, "current_load": current_load}, + json={"auth": GATEWAY_AUTH, "agent_id": agent_id, "current_load": current_load}, ) if resp.status_code == 200: return True @@ -93,7 +169,7 @@ def start_heartbeat(agent_id: str) -> threading.Thread: def unregister_agent(agent_id: str) -> bool: """向网关注销本 Agent。""" try: - resp = _client.post(f"{GATEWAY_URL}/api/agent/unregister", json={"agent_id": agent_id}) + resp = _client.post(f"{GATEWAY_URL}/api/agent/unregister", json={"auth": GATEWAY_AUTH, "agent_id": agent_id}) if resp.status_code in (200, 404): logger.info("unregistered from gateway agent=%s", agent_id) return True @@ -117,6 +193,7 @@ def report_result(request_id: str, agent_id: str, status: str = "success", error_info: 错误信息（失败时必填） """ body = { + "auth": GATEWAY_AUTH, "request_id": request_id, "agent_id": agent_id, "status": status, diff --git a/mcp_server/__init__.py b/mcp_server/__init__.py deleted file mode 100644 index 9fb2531..0000000 --- a/mcp_server/__init__.py +++ /dev/null @@ -1 +0,0 @@ -# mcp_server package diff --git a/mcp_server/agent_runner.py b/mcp_server/agent_runner.py deleted file mode 100644 index 9ef3903..0000000 --- a/mcp_server/agent_runner.py +++ /dev/null @@ -1,153 +0,0 @@ -""" -封装 ADK Agent 调用 -将 Dev Agent 的执行包装为可被任务管理器调用的异步函数 -""" -import os -import sys - -# 确保项目根目录在 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 google.adk.runners import Runner -from google.adk.sessions import InMemorySessionService -from google.genai import types - - -class AgentRunner: - def __init__(self): - # 延迟导入 agent，避免循环导入 - from agents.my_agent.agent import root_agent - self.agent = root_agent - self.session_service = InMemorySessionService() - self._runner: Runner = None - - def _get_runner(self) -> Runner: - if self._runner is None: - self._runner = Runner( - agent=self.agent, - app_name="dev_agent_server", - session_service=self.session_service, - auto_create_session=True, - ) - return self._runner - - async def run_task(self, task: dict) -> dict: - """ - 执行一个开发任务，返回结构化结果 - - Args: - task: 任务字典，包含 description, project_path, requirements 等 - - Returns: - 结构化的任务结果 - """ - task_id = task["id"] - description = task["description"] - project_path = task.get("project_path", "") - requirements = task.get("requirements", "") - - # 构建给 agent 的提示词 - prompt = self._build_prompt(description, project_path, requirements) - - print(f"[AgentRunner] 执行任务 {task_id}: {description[:60]}...") - - runner = self._get_runner() - session_id = f"task_{task_id}" - user_id = "task_manager" - - all_text = [] # 收集所有 model 消息中的文本 - tool_calls = [] # 收集所有 function_call - final_text = [] # 最终响应的文本 - event_count = 0 - error_msg = None - - try: - async for event in runner.run_async( - user_id=user_id, - session_id=session_id, - new_message=types.Content( - role="user", - parts=[types.Part(text=prompt)], - ), - ): - event_count += 1 - - # 跳过没有 content 的事件 - if not event.content: - continue - - # 只收集 model 角色的消息（不是 user / function_response 等） - content = event.content - role = getattr(content, "role", "") - if role != "model": - continue - - # 遍历 parts 收集文本和 function_call - parts = getattr(content, "parts", []) - for part in parts: - # 文本 - if hasattr(part, "text") and part.text: - all_text.append(part.text) - # 最终回复（没有 function_call 的 model 消息） - if event.is_final_response(): - final_text.append(part.text) - - # function_call - if hasattr(part, "function_call") and part.function_call: - fc = part.function_call - tool_calls.append({ - "name": fc.name, - "args": dict(fc.args) if hasattr(fc, "args") else {}, - }) - - result_text = "\n".join(final_text) if final_text else "\n".join(all_text) - status = "success" if result_text.strip() else "empty" - - except Exception as e: - error_msg = f"{type(e).__name__}: {e}" - result_text = f"任务执行出错：{error_msg}" - status = "error" - - return { - "summary": self._extract_summary(result_text), - "full_response": result_text, - "tool_calls_count": len(tool_calls), - "tool_calls_sample": tool_calls[:10], - "event_count": event_count, - "status": status, - "error": error_msg, - } - - def _build_prompt(self, description: str, project_path: str, requirements: str) -> str: - """构建给 agent 的任务指令""" - parts = [ - "你需要完成以下开发任务：", - "", - f"**任务描述：**{description}", - ] - if project_path: - parts.append(f"**项目路径：**{project_path}") - if requirements: - parts.append(f"**额外要求：**{requirements}") - parts.extend([ - "", - "请按照你的工作流程执行：", - "1. 浏览项目结构，理解上下文", - "2. 编写或修改代码", - "3. 运行编译/构建验证", - "4. 完成后，按照你规定的报告格式输出结果", - "", - "请开始执行。", - ]) - return "\n".join(parts) - - def _extract_summary(self, text: str) -> str: - """从 agent 回复中提取摘要""" - # 简单处理：取前 1000 字符作为摘要 - if not text: - return "(无响应)" - if len(text) <= 1000: - return text - return text[:1000] + "...（已截断）" diff --git a/mcp_server/server.py b/mcp_server/server.py deleted file mode 100644 index de7b5a1..0000000 --- a/mcp_server/server.py +++ /dev/null @@ -1,408 +0,0 @@ -""" -Dev Agent MCP Server（HTTP 模式） -将 Dev Agent 暴露为 MCP 服务器，CodeBuddy 通过 HTTP POST MCP 调用 - -直接用 FastAPI 处理 MCP JSON-RPC 消息，不依赖 mcp SDK 的 HTTP transport， -避免各种版本兼容问题。 -""" -import os -import sys -import asyncio -import json -import uuid -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 fastapi import FastAPI, Request, Response -import uvicorn - -from .task_manager import TaskManager -from .agent_runner import AgentRunner - - -# 加载 .env -load_dotenv(os.path.join(PROJECT_ROOT, "my_agent", ".env")) - -# 配置 -HOST = os.getenv("MCP_SERVER_HOST", "0.0.0.0") -PORT = int(os.getenv("MCP_SERVER_PORT", "8001")) -DATA_DIR = os.getenv("MCP_DATA_DIR", os.path.join(PROJECT_ROOT, "data")) -MCP_PATH = "/mcp" # MCP 端点路径 - -# 初始化组件 -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)) - - -# --- HTTP Server --- - -def create_fastapi_app() -> FastAPI: - """创建 FastAPI 应用，处理 MCP JSON-RPC 请求""" - fastapi_app = FastAPI(title="Dev Agent MCP Server") - - # 存储 session：session_id -> (read_stream, write_stream, session_task) - sessions = {} - - async def _get_or_create_session(session_id: str): - """获取或创建 MCP session（简单的内存会话管理）""" - if session_id not in sessions: - from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream - from mcp.server.session import ServerSession - - read_stream_writer, read_stream = MemoryObjectSendStream(100), MemoryObjectReceiveStream(100) - write_stream, write_stream_reader = MemoryObjectSendStream(100), MemoryObjectReceiveStream(100) - - session = ServerSession(read_stream, write_stream) - task = asyncio.create_task( - mcp_server.run( - read_stream, write_stream, - mcp_server.create_initialization_options(), - ) - ) - sessions[session_id] = (read_stream_writer, write_stream_reader, task, session) - - return sessions[session_id] - - @fastapi_app.post(MCP_PATH) - async def handle_mcp(request: Request): - """处理 MCP JSON-RPC 请求""" - body = await request.json() - - # 简单处理：单条请求（非批量） - # 从 header 获取 session_id，没有就创建新的 - session_id = request.headers.get("mcp-session-id") or str(uuid.uuid4()) - read_stream_writer, write_stream_reader, task, session = await _get_or_create_session(session_id) - - # 把请求写入 read_stream - await read_stream_writer.send(body) - - # 等待响应（简单地从 write_stream 读一条） - response = await write_stream_reader.receive() - - # 返回响应 - return Response( - content=json.dumps(response), - media_type="application/json", - headers={"mcp-session-id": session_id}, - ) - - @fastapi_app.get("/health") - async def health(): - return {"status": "ok", "server": "dev-agent-mcp-server"} - - return fastapi_app - - -async def _task_executor(task: dict) -> dict: - """任务执行器，交给 TaskManager 调用""" - return await agent_runner.run_task(task) - - -async def main(): - """启动 MCP Server""" - print("=" * 50) - print("Dev Agent MCP Server 启动中...") - print(f" 监听地址: {HOST}:{PORT}") - print(f" 数据目录: {DATA_DIR}") - print(f" MCP 端点: http://{HOST}:{PORT}{MCP_PATH}") - print(f" 健康检查: http://{HOST}:{PORT}/health") - print("=" * 50) - - # 启动任务管理器 - await task_manager.start(executor=_task_executor) - - # 启动 HTTP server - fastapi_app = create_fastapi_app() - config = uvicorn.Config(fastapi_app, host=HOST, port=PORT, log_level="info") - server = uvicorn.Server(config) - - try: - await server.serve() - finally: - await task_manager.stop() - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/mcp_server/start.py b/mcp_server/start.py deleted file mode 100644 index 49dd2dd..0000000 --- a/mcp_server/start.py +++ /dev/null @@ -1,25 +0,0 @@ -""" -Dev Agent MCP Server 启动脚本 -供 CodeBuddy stdio MCP 调用，使用绝对路径确保可以从任何目录启动 -""" -import os -import sys - -# 项目根目录（脚本所在目录的上一级） -PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) - -# 确保项目根目录在 Python path 里 -if PROJECT_ROOT not in sys.path: - sys.path.insert(0, PROJECT_ROOT) - -# 切换到项目目录，确保相对路径正确 -os.chdir(PROJECT_ROOT) - -# 强制 UTF-8 -os.environ["PYTHONUTF8"] = "1" - -# 导入并运行 -from mcp_server.stdio_server import main -import asyncio - -asyncio.run(main()) diff --git a/mcp_server/stdio_server.py b/mcp_server/stdio_server.py deleted file mode 100644 index ae1f1c8..0000000 --- a/mcp_server/stdio_server.py +++ /dev/null @@ -1,337 +0,0 @@ -""" -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()) diff --git a/mcp_server/task_manager.py b/mcp_server/task_manager.py deleted file mode 100644 index dc24eb0..0000000 --- a/mcp_server/task_manager.py +++ /dev/null @@ -1,164 +0,0 @@ -""" -异步任务管理器 -负责任务的提交、调度、状态管理 -""" -import asyncio -import uuid -import time -import os -from typing import Dict, List, Optional, Callable, Awaitable -from .task_store import TaskStore - - -# 任务状态 -STATUS_PENDING = "pending" -STATUS_RUNNING = "running" -STATUS_COMPLETED = "completed" -STATUS_FAILED = "failed" -STATUS_CANCELLED = "cancelled" - - -class TaskManager: - def __init__(self, store_dir: str = "./data/tasks", max_concurrent: int = 3): - self.store = TaskStore(store_dir) - self.max_concurrent = max_concurrent - self._tasks: Dict[str, dict] = {} - self._running = 0 - self._semaphore = asyncio.Semaphore(max_concurrent) - self._worker_task: Optional[asyncio.Task] = None - self._queue: asyncio.Queue = asyncio.Queue() - self._executor: Optional[Callable[[dict], Awaitable[dict]]] = None - - async def start(self, executor: Callable[[dict], Awaitable[dict]]): - """启动任务管理器，executor 是实际执行任务的异步函数""" - self._executor = executor - self._worker_task = asyncio.create_task(self._worker_loop()) - print(f"[TaskManager] 已启动，最大并发: {self.max_concurrent}") - - async def stop(self): - """停止任务管理器""" - if self._worker_task: - self._worker_task.cancel() - try: - await self._worker_task - except asyncio.CancelledError: - pass - print("[TaskManager] 已停止") - - async def submit_task(self, description: str, project_path: str, - requirements: str = "", extra: dict = None) -> dict: - """提交一个新任务""" - task_id = str(uuid.uuid4())[:8] - now = time.time() - task = { - "id": task_id, - "description": description, - "project_path": project_path, - "requirements": requirements, - "status": STATUS_PENDING, - "created_at": now, - "updated_at": now, - "result": None, - "logs": [], - "extra": extra or {}, - } - self._tasks[task_id] = task - self.store.save(task) - await self._queue.put(task_id) - print(f"[TaskManager] 任务已提交: {task_id} - {description[:50]}") - return task - - def get_task(self, task_id: str) -> Optional[dict]: - """获取任务详情""" - # 优先从内存取，没有再从文件读 - if task_id in self._tasks: - return self._tasks[task_id] - return self.store.load(task_id) - - def get_task_status(self, task_id: str) -> Optional[str]: - task = self.get_task(task_id) - return task["status"] if task else None - - def list_tasks(self, status: str = None) -> List[dict]: - """列出所有任务，可按状态过滤""" - tasks = list(self._tasks.values()) - # 加上磁盘上的任务 - disk_tasks = self.store.list_all() - disk_ids = {t["id"] for t in tasks} - for t in disk_tasks: - if t["id"] not in disk_ids: - tasks.append(t) - if status: - tasks = [t for t in tasks if t["status"] == status] - tasks.sort(key=lambda t: t.get("created_at", 0), reverse=True) - return tasks - - def append_log(self, task_id: str, message: str): - """追加任务日志""" - task = self._tasks.get(task_id) - if not task: - return - if "logs" not in task: - task["logs"] = [] - task["logs"].append({ - "time": time.strftime("%Y-%m-%d %H:%M:%S"), - "message": message, - }) - if len(task["logs"]) > 500: - task["logs"] = task["logs"][-500:] - # 异步持久化（这里直接同步写，简单起见） - self.store.save(task) - - async def cancel_task(self, task_id: str) -> bool: - """取消任务""" - task = self._tasks.get(task_id) - if not task: - return False - if task["status"] in (STATUS_COMPLETED, STATUS_FAILED, STATUS_CANCELLED): - return False - task["status"] = STATUS_CANCELLED - task["updated_at"] = time.time() - self.store.save(task) - print(f"[TaskManager] 任务已取消: {task_id}") - return True - - async def _worker_loop(self): - """后台 worker，从队列取任务执行""" - while True: - try: - task_id = await self._queue.get() - async with self._semaphore: - await self._execute_task(task_id) - except asyncio.CancelledError: - break - except Exception as e: - print(f"[TaskManager] Worker 异常: {e}") - await asyncio.sleep(1) - - async def _execute_task(self, task_id: str): - """执行单个任务""" - task = self._tasks.get(task_id) - if not task or task["status"] == STATUS_CANCELLED: - return - - task["status"] = STATUS_RUNNING - task["updated_at"] = time.time() - self.store.save(task) - print(f"[TaskManager] 开始执行: {task_id}") - - try: - result = await self._executor(task) - task["result"] = result - # 检查是否已被取消 - if task["status"] == STATUS_CANCELLED: - return - task["status"] = STATUS_COMPLETED - print(f"[TaskManager] 任务完成: {task_id}") - except Exception as e: - task["status"] = STATUS_FAILED - task["result"] = {"error": str(e)} - self.append_log(task_id, f"执行失败: {e}") - print(f"[TaskManager] 任务失败: {task_id} - {e}") - finally: - task["updated_at"] = time.time() - self.store.save(task) diff --git a/mcp_server/task_store.py b/mcp_server/task_store.py deleted file mode 100644 index b509f57..0000000 --- a/mcp_server/task_store.py +++ /dev/null @@ -1,56 +0,0 @@ -""" -任务持久化存储 -使用 JSON 文件存储任务数据 -""" -import os -import json -import time -from typing import Dict, List, Optional - - -class TaskStore: - def __init__(self, store_dir: str): - self.store_dir = os.path.abspath(store_dir) - os.makedirs(self.store_dir, exist_ok=True) - - def _task_path(self, task_id: str) -> str: - return os.path.join(self.store_dir, f"{task_id}.json") - - def save(self, task: dict) -> None: - task["updated_at"] = time.time() - path = self._task_path(task["id"]) - with open(path, "w", encoding="utf-8") as f: - json.dump(task, f, ensure_ascii=False, indent=2) - - def load(self, task_id: str) -> Optional[dict]: - path = self._task_path(task_id) - if not os.path.exists(path): - return None - with open(path, "r", encoding="utf-8") as f: - return json.load(f) - - def list_all(self) -> List[dict]: - tasks = [] - for filename in os.listdir(self.store_dir): - if filename.endswith(".json"): - task_id = filename[:-5] - task = self.load(task_id) - if task: - tasks.append(task) - tasks.sort(key=lambda t: t.get("created_at", 0), reverse=True) - return tasks - - def append_log(self, task_id: str, log_line: str) -> None: - task = self.load(task_id) - if not task: - return - if "logs" not in task: - task["logs"] = [] - task["logs"].append({ - "time": time.strftime("%Y-%m-%d %H:%M:%S"), - "message": log_line, - }) - # 日志最多保留 500 条 - if len(task["logs"]) > 500: - task["logs"] = task["logs"][-500:] - self.save(task) diff --git a/mcp_tools/__init__.py b/mcp_tools/__init__.py deleted file mode 100644 index 3675efb..0000000 --- a/mcp_tools/__init__.py +++ /dev/null @@ -1 +0,0 @@ -# mcp_tools package diff --git a/mcp_tools/command_executor/__init__.py b/mcp_tools/command_executor/__init__.py deleted file mode 100644 index f4b0530..0000000 --- a/mcp_tools/command_executor/__init__.py +++ /dev/null @@ -1 +0,0 @@ -# command_executor package diff --git a/mcp_tools/command_executor/server.py b/mcp_tools/command_executor/server.py deleted file mode 100644 index 1fa0d87..0000000 --- a/mcp_tools/command_executor/server.py +++ /dev/null @@ -1,127 +0,0 @@ -""" -终端命令执行 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()) \ No newline at end of file