HandsomeAq/ADK-agents
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

A2A工具初始化 测试通过

Browse Source
This commit is contained in:
handsomeAq 2026-07-29 17:21:34 +08:00
parent 7fe37768c4
commit bda9cb7782
11 changed files with 1443 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

224
PLAN.md Normal file
View File

@ -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 虚拟环境
```

53
a2a_server.py Normal file
View File

@ -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()

1
mcp_server/__init__.py Normal file
View File

@ -0,0 +1 @@
# mcp_server package

125
mcp_server/agent_runner.py Normal file
View File

@ -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] + "...（已截断）"

408
mcp_server/server.py Normal file
View File

@ -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())

25
mcp_server/start.py Normal file
View File

@ -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())

337
mcp_server/stdio_server.py Normal file
View File

@ -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())

164
mcp_server/task_manager.py Normal file
View File

@ -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)

56
mcp_server/task_store.py Normal file
View File

@ -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)

10
my_agent/agent.py
View File

@ -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"

43
test_sse_client.py Normal file
View File

@ -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())