From bda9cb7782be1634a31bd398257280dca3d56ea4 Mon Sep 17 00:00:00 2001 From: handsomeAq Date: Wed, 29 Jul 2026 17:21:34 +0800 Subject: [PATCH] =?UTF-8?q?A2A=E5=B7=A5=E5=85=B7=E5=88=9D=E5=A7=8B?= =?UTF-8?q?=E5=8C=96=20=E6=B5=8B=E8=AF=95=E9=80=9A=E8=BF=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- PLAN.md | 224 ++++++++++++++++++++ a2a_server.py | 53 +++++ mcp_server/__init__.py | 1 + mcp_server/agent_runner.py | 125 ++++++++++++ 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 +++++ my_agent/agent.py | 10 +- test_sse_client.py | 43 ++++ 11 files changed, 1443 insertions(+), 3 deletions(-) create mode 100644 PLAN.md create mode 100644 a2a_server.py create mode 100644 mcp_server/__init__.py create mode 100644 mcp_server/agent_runner.py create mode 100644 mcp_server/server.py create mode 100644 mcp_server/start.py create mode 100644 mcp_server/stdio_server.py create mode 100644 mcp_server/task_manager.py create mode 100644 mcp_server/task_store.py create mode 100644 test_sse_client.py diff --git a/PLAN.md b/PLAN.md new file mode 100644 index 0000000..09f1edb --- /dev/null +++ b/PLAN.md @@ -0,0 +1,224 @@ +# A2A 平台建设计划 + +> 基于 ADK + MCP 的主从式智能体平台 +> 主控：CodeBuddy ｜ 第一个子 Agent：全栈代码开发辅助 + +--- + +## 一、项目背景与目标 + +### 1.1 背景 +- 使用 ADK (Agent Development Kit) 搭建子智能体 +- 通过 LiteLlm 连接第三方 vLLM 端点：`https://9router.aqroid.cn/v1` +- 模型：`gikun1/gpt-5.6-luna`（OpenAI 兼容接口） +- **CodeBuddy 作为主控**，用户通过 CodeBuddy 调度子 agent 完成任务 + +### 1.2 目标 +搭建主从式 A2A (Agent-to-Agent) 平台，支持： +- 用户只跟 CodeBuddy 交互，由 CodeBuddy 调度子 agent 执行具体任务 +- 子 agent 通过 MCP HTTP 接口对外提供服务 +- 异步任务模式，支持长时任务执行 +- 可扩展多个子 agent（开发、测试、审查、运维等） + +### 1.3 典型工作流 + +``` +1. 用户向 CodeBuddy 描述需求（"给项目加个用户管理模块"） +2. CodeBuddy 拆解任务，制定方案 +3. 用户确认方案 +4. CodeBuddy 通过 MCP 调用子 Dev Agent 执行具体开发 +5. CodeBuddy 轮询任务进度，阶段性反馈给用户 +6. 子 agent 完成后，CodeBuddy 验收质量 +7. 有问题 → CodeBuddy 提修改意见 → 子 agent 修改 +8. 没问题 → 汇总结果给用户 +``` + +--- + +## 二、架构总览 + +``` +┌──────────────┐ +│ 用户 │ +└──────┬───────┘ + │ 自然语言交互 + ▼ +┌─────────────────────────────────────────────────┐ +│ CodeBuddy（主控） │ +│ • 理解需求 │ +│ • 拆解任务 & 制定方案 │ +│ • 调度子 agent │ +│ • 质量验收 & 结果汇总 │ +└───────────┬──────────────────┬──────────────────┘ + │ MCP HTTP 调用 │ MCP HTTP 调用 + ▼ ▼ +┌──────────────────┐ ┌──────────────────┐ +│ Dev Agent │ │ Test Agent │ ...更多子 agent +│ (全栈代码开发) │ │ (测试编写执行) │ +│ - ADK + LiteLlm │ │ - ADK + LiteLlm │ +│ - MCP Server │ │ - MCP Server │ +└────────┬─────────┘ └────────┬─────────┘ + │ 内部工具 │ 内部工具 + ▼ ▼ + ┌─────────┐ ┌─────────┐ + │ 文件系统 │ │ 终端 │ + │ Git │ │ 构建命令 │ + └─────────┘ └─────────┘ +``` + +--- + +## 三、阶段规划 + +### 阶段一：加固代码辅助 Agent ✅ + +**目标：** 让 Dev Agent 具备完整的全栈代码开发能力（内部能力建设） + +#### 1.1 接入文件系统工具 ✅ +- 使用 `@modelcontextprotocol/server-filesystem`（Node.js MCP server） +- 能力：读取文件、写入文件、列出目录、搜索文件 +- 配置：`StdioConnectionParams` + 30s 超时 + +#### 1.2 接入终端命令执行工具 ✅ +- 实现方式：Python 原生 `FunctionTool` + `asyncio.create_subprocess_shell` +- 最初尝试 MCP server 但 Windows 下通信不稳定，改为直接函数工具 +- 能力：执行 shell 命令（编译、构建、测试等），支持超时控制和输出截断 + +#### 1.3 优化 Agent 指令与角色 ✅ +- 明确身份：全栈开发子 Agent，接受 CodeBuddy 调度 +- 四大模块：工作流程 / 工作边界 / 编译构建守则 / 报告格式 +- 上报机制：架构决策、依赖不兼容、环境问题、能力范围外 +- 保护机制：连续修复 3 次仍失败则上报 + +**涉及文件：** +``` +my_agent/ +├── agent.py # Dev Agent 主文件 +├── .env # 环境配置 +└── __init__.py +mcp_tools/ +└── command_executor/ # （备用，当前未使用） + └── server.py +``` + +--- + +### 阶段二：Dev Agent 暴露为 MCP Server（HTTP/SSE） + +**目标：** 将 Dev Agent 包装为 MCP 服务器，CodeBuddy 可通过 HTTP 调用 + +#### 2.1 MCP Server（SSE 模式） +- 使用 `mcp` Python SDK 的 SSE server +- 通信协议：HTTP + Server-Sent Events +- 端口：可配置（默认 8001） + +#### 2.2 异步任务接口 +暴露以下 MCP 工具： + +| 工具名 | 描述 | 参数 | +|--------|------|------| +| `submit_task` | 提交开发任务 | `description: str`, `project_path: str`, `requirements?: str` | +| `get_task_status` | 查询任务状态 | `task_id: str` | +| `get_task_result` | 获取任务结果 | `task_id: str` | +| `get_task_log` | 获取任务执行日志 | `task_id: str` | +| `cancel_task` | 取消任务 | `task_id: str` | +| `list_tasks` | 列出任务列表 | `status?: str` | + +任务状态流转：`pending` → `running` → `completed` / `failed` + +#### 2.3 任务管理器 +- 内存中的任务队列（支持并发） +- 任务日志持久化（文件存储） +- 任务结果结构化返回 + +#### 2.4 CodeBuddy 接入验证 +- CodeBuddy 配置 MCP 客户端连接 Dev Agent +- 验证：提交任务 → 轮询进度 → 获取结果 → 验收 + +**涉及文件：** +``` +mcp_server/ +├── __init__.py +├── server.py # MCP SSE server 主入口 +├── task_manager.py # 异步任务管理器 +├── agent_runner.py # 封装 ADK agent 调用 +└── task_store.py # 任务持久化存储 +``` + +--- + +### 阶段三：完善协作流程 + +**目标：** 优化 CodeBuddy 与子 agent 的协作效率 + +#### 3.1 任务描述规范 +- 定义标准化的任务提交格式 +- 包含：需求描述、项目路径、技术约束、验收标准 + +#### 3.2 增量修改支持 +- 支持基于已有修改继续迭代（CodeBuddy 提修改意见 → agent 修改） +- 任务关联与版本管理 + +#### 3.3 错误处理与重试 +- 子 agent 失败时的错误报告格式 +- CodeBuddy 端的重试策略 + +--- + +### 阶段四：扩展更多子 Agent（远期） + +| Agent | 职责 | 主要工具 | +|-------|------|----------| +| Dev Agent | 全栈代码开发 | 文件系统、终端、Git | +| Test Agent | 测试编写与执行 | 终端、覆盖率工具 | +| Review Agent | 代码审查 | 文件系统、静态分析 | +| DevOps Agent | 部署与运维 | 云平台、CI/CD | + +--- + +## 四、技术栈 + +| 层级 | 技术 | +|------|------| +| 主控 | CodeBuddy（对话式调度） | +| 子 Agent 框架 | Google ADK (Python) | +| LLM 接入 | LiteLlm → vLLM 兼容端点 | +| MCP | `mcp` Python SDK（官方） | +| Agent 内部工具 | 文件系统 MCP Server + Python 原生 FunctionTool | +| 任务存储 | 本地 JSON 文件（初期） | +| 通信协议 | HTTP/SSE (MCP) | + +--- + +## 五、实施路线 + +| 阶段 | 内容 | 状态 | +|------|------|------| +| 阶段一 | Dev Agent + 文件系统 + 终端工具 | ✅ 完成 | +| 阶段二 | MCP Server + 异步任务 | 🔄 进行中 | +| 阶段三 | 协作流程优化 | ⏳ 待开始 | +| 阶段四 | 扩展子 agent | ⏳ 待开始 | + +--- + +## 六、项目结构 + +``` +agent/ +├── PLAN.md # 本文件 +├── my_agent/ # Dev Agent +│ ├── agent.py # Agent 定义（含文件系统 MCP + run_command 工具） +│ ├── .env # 环境变量 +│ └── __init__.py +├── mcp_server/ # MCP HTTP Server（阶段二） +│ ├── server.py # MCP SSE server 主入口 +│ ├── task_manager.py # 异步任务管理器 +│ ├── agent_runner.py # 封装 ADK agent 调用 +│ ├── task_store.py # 任务持久化存储 +│ └── __init__.py +├── mcp_tools/ # 自定义 MCP 工具（备用） +│ └── command_executor/ +│ ├── server.py +│ └── __init__.py +└── .venv/ # Python 虚拟环境 +``` diff --git a/a2a_server.py b/a2a_server.py new file mode 100644 index 0000000..354ce0d --- /dev/null +++ b/a2a_server.py @@ -0,0 +1,53 @@ +""" +Dev Agent A2A Server +使用 ADK 官方 A2A 协议暴露 Dev Agent，支持 HTTP 调用，可直接上云部署。 + +启动方式： + python a2a_server.py + +端点： + A2A 接口: http://127.0.0.1:8001/a2a/dev_agent + Agent卡片: http://127.0.0.1:8001/.well-known/agent-card.json + 健康检查: http://127.0.0.1:8001/health +""" +import os +import sys + +# 确保项目根目录在 path 里 +PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__)) +if PROJECT_ROOT not in sys.path: + sys.path.insert(0, PROJECT_ROOT) + +from dotenv import load_dotenv +load_dotenv(os.path.join(PROJECT_ROOT, "my_agent", ".env")) + +# 强制 UTF-8 +os.environ["PYTHONUTF8"] = "1" + +import uvicorn +from google.adk.a2a.utils.agent_to_a2a import to_a2a +from my_agent.agent import root_agent + + +# 配置 +HOST = os.getenv("A2A_SERVER_HOST", "0.0.0.0") +PORT = int(os.getenv("A2A_SERVER_PORT", "8001")) + +# 用 ADK 官方工具把 agent 转成 A2A 服务 +# 会自动生成 agent card，暴露 /a2a/{agent_name} 端点 +a2a_app = to_a2a(root_agent, port=PORT) + + +def main(): + print("=" * 60) + print("Dev Agent A2A Server 启动中...") + print(f" 监听地址: http://{HOST}:{PORT}") + print(f" A2A 端点: http://{HOST}:{PORT}/a2a/{root_agent.name}") + print(f" Agent 卡片: http://{HOST}:{PORT}/.well-known/agent-card.json") + print("=" * 60) + + uvicorn.run(a2a_app, host=HOST, port=PORT, log_level="info") + + +if __name__ == "__main__": + main() diff --git a/mcp_server/__init__.py b/mcp_server/__init__.py new file mode 100644 index 0000000..9fb2531 --- /dev/null +++ b/mcp_server/__init__.py @@ -0,0 +1 @@ +# mcp_server package diff --git a/mcp_server/agent_runner.py b/mcp_server/agent_runner.py new file mode 100644 index 0000000..c1de72d --- /dev/null +++ b/mcp_server/agent_runner.py @@ -0,0 +1,125 @@ +""" +封装 ADK Agent 调用 +将 Dev Agent 的执行包装为可被任务管理器调用的异步函数 +""" +import os +import sys +import asyncio + +# 确保项目根目录在 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 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 = [] + tool_calls = [] + + 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)], + ), + ): + # 收集文本输出 + if hasattr(event, 'output') and event.output: + content = event.output + if hasattr(content, 'parts'): + for part in content.parts: + if hasattr(part, 'text') and part.text: + all_text.append(part.text) + if hasattr(part, 'function_call') and part.function_call: + tool_calls.append({ + "name": part.function_call.name, + "args": dict(part.function_call.args) if hasattr(part.function_call, 'args') else {}, + }) + + result_text = "".join(all_text) + + return { + "summary": self._extract_summary(result_text), + "full_response": result_text, + "tool_calls_count": len(tool_calls), + "tool_calls_sample": tool_calls[:10], # 只保留前 10 个 + "status": "success" if result_text else "empty", + } + + 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 new file mode 100644 index 0000000..de7b5a1 --- /dev/null +++ b/mcp_server/server.py @@ -0,0 +1,408 @@ +""" +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 new file mode 100644 index 0000000..49dd2dd --- /dev/null +++ b/mcp_server/start.py @@ -0,0 +1,25 @@ +""" +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 new file mode 100644 index 0000000..ae1f1c8 --- /dev/null +++ b/mcp_server/stdio_server.py @@ -0,0 +1,337 @@ +""" +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 new file mode 100644 index 0000000..dc24eb0 --- /dev/null +++ b/mcp_server/task_manager.py @@ -0,0 +1,164 @@ +""" +异步任务管理器 +负责任务的提交、调度、状态管理 +""" +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 new file mode 100644 index 0000000..b509f57 --- /dev/null +++ b/mcp_server/task_store.py @@ -0,0 +1,56 @@ +""" +任务持久化存储 +使用 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/my_agent/agent.py b/my_agent/agent.py index 6e36801..e6541c7 100644 --- a/my_agent/agent.py +++ b/my_agent/agent.py @@ -19,7 +19,6 @@ api_key = os.getenv("VLLM_API_KEY", "") # Agent 可访问的工作目录 WORKSPACE_DIR = os.getenv("AGENT_WORKSPACE_DIR", r"D:\nzy\workspace_git") - # --- 文件系统 MCP 工具 --- filesystem_mcp = McpToolset( connection_params=StdioConnectionParams( @@ -95,7 +94,6 @@ async def run_command(command: str, cwd: str = None, timeout: int = 300) -> str: # 注册为 ADK 工具 run_command_tool = FunctionTool(run_command) - root_agent = LlmAgent( model=LiteLlm( model=model_name, @@ -106,7 +104,13 @@ root_agent = LlmAgent( name="dev_agent", description="全栈开发子 Agent，可以读写文件、浏览目录、执行开发任务。", instruction=( - "你是一个全栈开发子 Agent（Dev Agent），由主控调度执行具体的开发任务。\n" + "姓名：花花✿\n" + "性别：女\n" + "籍贯：东北（辽宁沈阳），地道东北姑娘，26 岁\n" + "职业：资深全栈开发工程师，5 年一线企业级开发经验\n" + "外形气质：性格爽朗大方，说话直爽不绕弯，共情力强；做事严谨较真，技术上极度靠谱，\n" + "生活里接地气、热心肠；不矫情，能扛项目压力，也会温柔安抚焦虑的开发同事\n" + "由主控调度执行具体的开发任务。\n" "\n" "## 工作流程\n" "1. 先理解任务需求和项目上下文\n" diff --git a/test_sse_client.py b/test_sse_client.py new file mode 100644 index 0000000..b953783 --- /dev/null +++ b/test_sse_client.py @@ -0,0 +1,43 @@ +"""测试 SSE MCP 客户端连接""" +import asyncio +import sys + +from mcp.client.sse import sse_client +from mcp.client.session import ClientSession + + +async def main(): + url = "http://127.0.0.1:8001/mcp/sse" + print(f"连接到: {url}") + + try: + async with sse_client(url) as (read_stream, write_stream): + print("SSE 连接建立成功！") + + async with ClientSession(read_stream, write_stream) as session: + print("正在初始化 MCP 会话...") + result = await session.initialize() + print(f"初始化成功! Server: {result.serverInfo.name} v{result.serverInfo.version}") + + print("\n列出工具...") + tools = await session.list_tools() + print(f"工具数量: {len(tools.tools)}") + for t in tools.tools: + print(f" - {t.name}: {t.description[:60]}") + + print("\n提交一个测试任务...") + result = await session.call_tool("submit_task", { + "description": "这是一个测试任务，请回复 hello world", + "project_path": "D:\\nzy\\workspace_git", + }) + print(f"结果: {result}") + + except Exception as e: + print(f"连接失败: {type(e).__name__}: {e}") + import traceback + traceback.print_exc() + sys.exit(1) + + +if __name__ == "__main__": + asyncio.run(main())