139 lines
4.2 KiB
Python
139 lines
4.2 KiB
Python
"""
|
||
Dev Agent MCP Server
|
||
将 Dev Agent 封装为 MCP 工具，供 CodeBuddy 等 MCP 客户端调用。
|
||
|
||
功能：
|
||
- run_dev_agent: 提交任务给 Dev Agent 执行，返回执行结果
|
||
- 支持指定 session_id 进行多轮对话
|
||
- 自动提取最终回复文本
|
||
|
||
环境变量配置：
|
||
- DEV_AGENT_API_URL: API Server 地址（默认 http://127.0.0.1:8001）
|
||
- DEV_AGENT_APP_NAME: 调用的 agent 名称（默认 my_agent）
|
||
- DEV_AGENT_USER_ID: 用户 ID（默认 codebuddy）
|
||
"""
|
||
import os
|
||
import sys
|
||
|
||
# 强制 UTF-8
|
||
os.environ["PYTHONUTF8"] = "1"
|
||
|
||
import httpx
|
||
from mcp.server.fastmcp import FastMCP
|
||
|
||
|
||
# 配置
|
||
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", "my_agent")
|
||
DEV_AGENT_USER_ID = os.getenv("DEV_AGENT_USER_ID", "codebuddy")
|
||
|
||
# MCP Server
|
||
mcp = FastMCP(name=f"dev-agent-{DEV_AGENT_APP_NAME}")
|
||
|
||
|
||
@mcp.tool()
|
||
async def run_dev_agent(task: str, session_id: str = "default") -> str:
|
||
"""调用 Dev Agent（全栈开发子 Agent）执行开发任务。
|
||
|
||
Dev Agent 可以读写文件、运行终端命令、执行编译/构建/测试。
|
||
适用于：代码编写、bug 修复、项目搭建、编译验证等开发子任务。
|
||
|
||
Args:
|
||
task: 要 Dev Agent 执行的任务描述，越详细越好
|
||
session_id: 可选，会话 ID，用于多轮对话/续聊，默认 "default"
|
||
|
||
Returns:
|
||
Dev Agent 的执行结果
|
||
"""
|
||
if not task:
|
||
return "错误：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 (
|
||
f"调用 Dev Agent 失败（HTTP {response.status_code}）:\n"
|
||
f"{response.text[:500]}"
|
||
)
|
||
|
||
events = response.json()
|
||
except httpx.ConnectError:
|
||
return (
|
||
f"无法连接到 Dev Agent API Server（{DEV_AGENT_API_URL}）\n"
|
||
f"请确认 api_server.py 是否已启动。"
|
||
)
|
||
except Exception as e:
|
||
return f"调用 Dev Agent 出错: {e}"
|
||
|
||
# 从事件列表中提取最终回复
|
||
return _extract_final_response(events, session_id)
|
||
|
||
|
||
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
|
||
|
||
|
||
def main():
|
||
"""stdio 模式启动 MCP server"""
|
||
# Windows stdio 二进制模式
|
||
if sys.platform == "win32":
|
||
import msvcrt
|
||
msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)
|
||
msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)
|
||
|
||
mcp.run(transport="stdio")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|