184 lines
5.8 KiB
Python
184 lines
5.8 KiB
Python
"""
|
||
Dev Agent MCP Server
|
||
将 Dev Agent 封装为 MCP 工具，供 CodeBuddy 等 MCP 客户端调用。
|
||
|
||
功能：
|
||
- run_dev_agent: 提交任务给 Dev Agent 执行，返回执行结果
|
||
- 支持指定 session_id 进行多轮对话
|
||
- 自动提取最终回复文本
|
||
|
||
启动方式（MCP 配置）：
|
||
{
|
||
"mcpServers": {
|
||
"dev-agent": {
|
||
"command": "python",
|
||
"args": ["d:/nzy/workspace_python/agent/mcp_dev_agent/server.py"],
|
||
"env": {
|
||
"DEV_AGENT_API_URL": "http://127.0.0.1:8001",
|
||
"DEV_AGENT_APP_NAME": "dev_agent",
|
||
"DEV_AGENT_USER_ID": "codebuddy"
|
||
}
|
||
}
|
||
}
|
||
}
|
||
"""
|
||
import os
|
||
import json
|
||
import asyncio
|
||
import sys
|
||
from typing import Any
|
||
|
||
import httpx
|
||
from mcp.server import Server
|
||
from mcp.types import Tool, TextContent
|
||
from mcp.server.stdio import stdio_server
|
||
|
||
|
||
# 配置
|
||
DEV_AGENT_API_URL = os.getenv("DEV_AGENT_API_URL", "http://127.0.0.1:8001")
|
||
DEV_AGENT_APP_NAME = os.getenv("DEV_AGENT_APP_NAME", "dev_agent")
|
||
DEV_AGENT_USER_ID = os.getenv("DEV_AGENT_USER_ID", "codebuddy")
|
||
|
||
# MCP Server
|
||
server = Server("dev-agent-mcp")
|
||
|
||
|
||
@server.list_tools()
|
||
async def list_tools() -> list[Tool]:
|
||
"""列出可用工具"""
|
||
return [
|
||
Tool(
|
||
name="run_dev_agent",
|
||
description=(
|
||
"调用 Dev Agent（全栈开发子 Agent）执行开发任务。\n"
|
||
"Dev Agent 可以读写文件、运行终端命令、执行编译/构建/测试。\n"
|
||
"适用于：代码编写、bug 修复、项目搭建、编译验证等开发子任务。\n"
|
||
"传参说明：task 是任务描述，session_id 可选，不传则创建新会话。"
|
||
),
|
||
inputSchema={
|
||
"type": "object",
|
||
"properties": {
|
||
"task": {
|
||
"type": "string",
|
||
"description": "要 Dev Agent 执行的任务描述，越详细越好",
|
||
},
|
||
"session_id": {
|
||
"type": "string",
|
||
"description": "可选，会话 ID，用于多轮对话/续聊",
|
||
},
|
||
},
|
||
"required": ["task"],
|
||
},
|
||
),
|
||
]
|
||
|
||
|
||
@server.call_tool()
|
||
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
|
||
"""调用工具"""
|
||
if name == "run_dev_agent":
|
||
return await _run_dev_agent(arguments)
|
||
else:
|
||
raise ValueError(f"Unknown tool: {name}")
|
||
|
||
|
||
async def _run_dev_agent(args: dict[str, Any]) -> list[TextContent]:
|
||
"""调用 Dev Agent API 执行任务"""
|
||
task = args.get("task", "").strip()
|
||
session_id = args.get("session_id") or "default"
|
||
|
||
if not task:
|
||
return [TextContent(type="text", text="错误：task 不能为空")]
|
||
|
||
# 构造请求
|
||
payload = {
|
||
"appName": DEV_AGENT_APP_NAME,
|
||
"userId": DEV_AGENT_USER_ID,
|
||
"sessionId": session_id,
|
||
"newMessage": {
|
||
"role": "user",
|
||
"parts": [{"text": task}],
|
||
},
|
||
}
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=600.0) as client: # 10 分钟超时
|
||
response = await client.post(
|
||
f"{DEV_AGENT_API_URL}/run",
|
||
json=payload,
|
||
headers={"Content-Type": "application/json"},
|
||
)
|
||
|
||
if response.status_code != 200:
|
||
return [TextContent(
|
||
type="text",
|
||
text=f"调用 Dev Agent 失败（HTTP {response.status_code}）:\n{response.text[:500]}"
|
||
)]
|
||
|
||
events = response.json()
|
||
except httpx.ConnectError:
|
||
return [TextContent(
|
||
type="text",
|
||
text=f"无法连接到 Dev Agent API Server（{DEV_AGENT_API_URL}）\n"
|
||
f"请确认 api_server.py 是否已启动。"
|
||
)]
|
||
except Exception as e:
|
||
return [TextContent(type="text", text=f"调用 Dev Agent 出错: {e}")]
|
||
|
||
# 从事件列表中提取最终回复
|
||
result_text = _extract_final_response(events, session_id)
|
||
return [TextContent(type="text", text=result_text)]
|
||
|
||
|
||
def _extract_final_response(events: list[dict], session_id: str) -> str:
|
||
"""从事件列表中提取 agent 的最终文本回复"""
|
||
if not events:
|
||
return "（无返回事件）"
|
||
|
||
final_text_parts = []
|
||
|
||
for event in events:
|
||
content = event.get("content", {})
|
||
role = content.get("role", "")
|
||
parts = content.get("parts", [])
|
||
author = event.get("author", "")
|
||
|
||
if role == "model" and author == DEV_AGENT_APP_NAME:
|
||
for part in parts:
|
||
if "text" in part:
|
||
final_text_parts.append(part["text"])
|
||
|
||
response = "\n".join(final_text_parts).strip()
|
||
|
||
if not response:
|
||
# 如果没有找到最终回复，返回事件摘要
|
||
summary = f"共 {len(events)} 个事件\n"
|
||
for e in events[-5:]:
|
||
content = e.get("content", {})
|
||
parts = content.get("parts", [])
|
||
role = content.get("role", "")
|
||
author = e.get("author", "")
|
||
part_types = [list(p.keys())[0] for p in parts]
|
||
summary += f" - [{role}] {author}: {part_types}\n"
|
||
response = f"（未提取到最终文本回复）\n{summary}"
|
||
|
||
# 附上 session_id 方便续聊
|
||
response += f"\n\n---\nsession_id: {session_id}"
|
||
return response
|
||
|
||
|
||
async def main():
|
||
"""stdio 模式启动 MCP server"""
|
||
async with stdio_server() as (read_stream, write_stream):
|
||
await server.run(read_stream, write_stream, None)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 确保 Windows 下 stdio 用二进制模式
|
||
if sys.platform == "win32":
|
||
import msvcrt
|
||
msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)
|
||
msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)
|
||
|
||
asyncio.run(main())
|