ADK-agents/mcp_dev_agent/server.py

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