AI Agent互操作性实战：用MCP和A2A打通不同Agent框架

🏗️ 打通不同框架的Agent，就像让说中文的人和说英文的人合作——协议是翻译官，但真正干活还得靠架构。

symptom_zh

上篇讲了MCP和A2A的理论区别。这篇全是代码和架构——Cline + LangChain + 自建Agent怎么通过MCP共享工具，怎么通过A2A互相委托任务。 搭建一个真实的多Agent协作系统，让不同框架的Agent能互相通信、共享工具、分工协作。包含完整的Docker Compose部署方案。

适合人群： 需要在生产环境搭建多Agent系统的开发者 读完能做： 搭建自己的多Agent协作系统

diagnosis_zh

一、架构总览

我们搭建一个生产级的多Agent系统，包含三种不同类型的Agent：

┌──────────────────────────────────────────────────────┐
│                  A2A 协议层（HTTP）                    │
│    ┌──────────┐    ┌──────────┐    ┌──────────┐     │
│    │  Orchestrator  │  Cline Agent   │  LangChain    │
│    │  Agent    │    │  (代码)    │    │  (分析)    │     │
│    └────┬─────┘    └────┬─────┘    └────┬─────┘     │
│         │               │               │           │
└─────────┼───────────────┼───────────────┼───────────┘
          │               │               │
          └───────────────┼───────────────┘
                          │
                    ┌─────▼─────┐
                    │  MCP 总线  │
                    └──┬──┬──┬──┘
          ┌─────────────┼──┼──┼─────────────┐
          ▼             ▼  ▼  ▼             ▼
    ┌──────────┐  ┌────────┐ ┌────────┐ ┌────────┐
    │数据库MCP │  │搜索MCP │ │文件MCP │ │SlackMCP│
    └──────────┘  └────────┘ └────────┘ └────────┘

设计原则： 1. 所有工具通过MCP总线共享——写一次MCP Server，所有Agent都能用 2. Agent之间通过A2A协议通信——框架无关 3. 一个中央Orchestrator Agent负责任务分发

二、搭建MCP总线的完整代码

MCP Server：搜索工具

# search_mcp_server.py
import json
import asyncio
import httpx
from mcp.server import Server
from mcp.server.stdio import StdioServerTransport
from mcp.types import Tool, TextContent

class SearchMCPServer:
    """通用搜索MCP服务器"""

    def __init__(self):
        self.serp_api_key = "your_key_here"
        self.tavily_api_key = "your_key_here"

    async def search(self, query: str, source: str = "web") -> str:
        """执行搜索"""
        if source == "web":
            async with httpx.AsyncClient() as client:
                resp = await client.post(
                    "https://api.tavily.com/search",
                    json={"query": query, "api_key": self.tavily_api_key}
                )
                data = resp.json()
                return json.dumps(data["results"][:5], ensure_ascii=False)
        return json.dumps({"error": "unsupported source"})

    async def run(self):
        app = Server("search-server")

        @app.list_tools()
        async def list_tools():
            return [
                Tool(
                    name="web_search",
                    description="搜索互联网信息，返回Top 5结果",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "query": {
                                "type": "string",
                                "description": "搜索关键词"
                            }
                        },
                        "required": ["query"]
                    }
                )
            ]

        @app.call_tool()
        async def call_tool(name: str, arguments: dict):
            if name == "web_search":
                result = await self.search(arguments["query"])
                return [TextContent(type="text", text=result)]
            raise ValueError(f"Unknown tool: {name}")

        transport = StdioServerTransport()
        await app.connect(transport)
        await app.wait_for_shutdown()

if __name__ == "__main__":
    asyncio.run(SearchMCPServer().run())

MCP Server：文件系统操作

# filesystem_mcp_server.py
import os
import json
from mcp.server import Server
from mcp.server.stdio import StdioServerTransport
from mcp.types import Tool, TextContent

class FileSystemMCPServer:
    """文件系统操作MCP服务器（安全受限版）"""

    def __init__(self, allowed_dirs: list[str]):
        self.allowed_dirs = [os.path.abspath(d) for d in allowed_dirs]

    def _check_path(self, path: str) -> str:
        abs_path = os.path.abspath(path)
        for allowed in self.allowed_dirs:
            if abs_path.startswith(allowed):
                return abs_path
        raise PermissionError(f"无权访问: {path}")

    async def run(self):
        app = Server("filesystem-server")

        @app.list_tools()
        async def list_tools():
            return [
                Tool(
                    name="read_file",
                    description="读取文件内容",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "path": {"type": "string", "description": "文件路径"}
                        },
                        "required": ["path"]
                    }
                ),
                Tool(
                    name="list_directory",
                    description="列出目录内容",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "path": {"type": "string", "description": "目录路径"}
                        },
                        "required": ["path"]
                    }
                )
            ]

        @app.call_tool()
        async def call_tool(name: str, arguments: dict):
            safe_path = self._check_path(arguments["path"])

            if name == "read_file":
                with open(safe_path, "r", encoding="utf-8") as f:
                    content = f.read()
                return [TextContent(type="text", text=content[:5000])]

            elif name == "list_directory":
                items = os.listdir(safe_path)
                return [TextContent(type="text", text=json.dumps(items, ensure_ascii=False))]

        transport = StdioServerTransport()
        await app.connect(transport)
        await app.wait_for_shutdown()

if __name__ == "__main__":
    import asyncio
    server = FileSystemMCPServer(allowed_dirs=["/data/workspace"])
    asyncio.run(server.run())

三、A2A Agent通信层

A2A Agent基类

# a2a_protocol.py
import json
import httpx
from typing import Optional
from dataclasses import dataclass, asdict

@dataclass
class AgentCard:
    """Agent的能力声明（A2A Agent Card）"""
    name: str
    description: str
    url: str
    skills: list[dict]

@dataclass
class TaskResult:
    """任务执行结果"""
    task_id: str
    status: str  # pending, running, completed, failed
    result: Optional[str] = None
    error: Optional[str] = None

class A2AAgent:
    """A2A协议的Agent基类"""

    def __init__(self, name: str, description: str, port: int):
        self.name = name
        self.description = description
        self.port = port
        self.card = AgentCard(
            name=name,
            description=description,
            url=f"http://localhost:{port}",
            skills=[]  # 子类实现
        )

    async def handle_task(self, task: dict) -> dict:
        """处理任务（子类实现）"""
        raise NotImplementedError

    async def call_agent(self, target_url: str, task: dict) -> TaskResult:
        """通过A2A协议调用另一个Agent"""
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{target_url}/a2a/task",
                json={"task": task},
                timeout=30.0
            )
            result = resp.json()
            # 如果任务还在跑，轮询结果
            while result["status"] in ("pending", "running"):
                await asyncio.sleep(1)
                resp = await client.get(f"{target_url}/a2a/task/{result['task_id']}")
                result = resp.json()
            return TaskResult(**result)

具体A2A Agent实现：分析Agent

# analyzer_agent.py（A2A Agent）
from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn
from a2a_protocol import A2AAgent, TaskResult

class TaskRequest(BaseModel):
    task: dict

class AnalysisAgent(A2AAgent):
    """数据分析Agent - 通过A2A对外暴露能力"""

    def __init__(self):
        super().__init__(
            name="data-analyzer",
            description="分析数据并提供洞察",
            port=8081
        )
        self.card.skills = [
            {"id": "analyze", "name": "数据分析", 
             "description": "对提供的数据进行分析并返回洞察"}
        ]

    async def handle_task(self, task: dict) -> dict:
        skill = task.get("skill", "analyze")
        params = task.get("params", {})

        if skill == "analyze":
            data = params.get("data", "")
            analysis = self._analyze(data)
            return {
                "task_id": task.get("task_id", "unknown"),
                "status": "completed",
                "result": analysis
            }

        return {"task_id": task.get("task_id", "unknown"), "status": "failed", "error": "Unknown skill"}

    def _analyze(self, data: str) -> str:
        # 使用MCP搜索工具找背景信息，然后分析
        # （实际代码中会调用MCP Client）
        return f"分析结果：数据包含{len(data)}个字符，建议进一步处理..."

# FastAPI应用
app = FastAPI()
agent = AnalysisAgent()

@app.post("/a2a/task")
async def receive_task(req: TaskRequest):
    return await agent.handle_task(req.task)

@app.get("/a2a/task/{task_id}")
async def get_task(task_id: str):
    return {"task_id": task_id, "status": "completed"}

@app.get("/.well-known/agent.json")
async def get_agent_card():
    return agent.card

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8081)

四、Orchestrator：任务分发中枢

# orchestrator.py - 核心控制
import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from a2a_protocol import A2AAgent

class Orchestrator:
    """Orchestrator Agent：负责任务分发和MCP工具调度"""

    def __init__(self):
        self.mcp_clients = {}
        self.a2a_agents = {}

    async def register_mcp(self, name: str, command: str, args: list[str]):
        """注册MCP Server"""
        server_params = StdioServerParameters(command=command, args=args)
        read, write = await stdio_client(server_params).__aenter__()
        session = await ClientSession(read, write).__aenter__()
        await session.initialize()
        self.mcp_clients[name] = session
        print(f"✅ MCP Server [{name}] 已注册")

    async def register_a2a(self, name: str, url: str):
        """注册A2A Agent"""
        import httpx
        async with httpx.AsyncClient() as client:
            resp = await client.get(f"{url}/.well-known/agent.json")
            card = resp.json()
            self.a2a_agents[name] = {"url": url, "card": card}
            print(f"✅ A2A Agent [{name}] 已注册: {card['description']}")

    async def call_mcp_tool(self, server_name: str, tool_name: str, arguments: dict):
        """调用MCP工具"""
        session = self.mcp_clients.get(server_name)
        if not session:
            return {"error": f"Unknown MCP server: {server_name}"}
        result = await session.call_tool(tool_name, arguments)
        return {"server": server_name, "tool": tool_name, "result": result.content[0].text}

    async def delegate_to_agent(self, agent_name: str, task: dict):
        """委托任务给A2A Agent"""
        agent_info = self.a2a_agents.get(agent_name)
        if not agent_info:
            return {"error": f"Unknown agent: {agent_name}"}

        import httpx
        async with httpx.AsyncClient() as client:
            resp = await client.post(
                f"{agent_info['url']}/a2a/task",
                json={"task": {"task_id": task.get("id"), "skill": task.get("skill"), "params": task.get("params")}},
                timeout=30.0
            )
            return resp.json()

    async def execute_workflow(self, workflow: list[dict]) -> list[dict]:
        """执行工作流"""
        results = []
        for step in workflow:
            if step["type"] == "mcp":
                result = await self.call_mcp_tool(
                    step["server"], step["tool"], step.get("args", {})
                )
            elif step["type"] == "a2a":
                result = await self.delegate_to_agent(
                    step["agent"], step["task"]
                )
            else:
                result = {"error": f"Unknown step type: {step['type']}"}

            results.append({"step": step["name"], "result": result})
        return results

# 使用示例
async def main():
    orchestrator = Orchestrator()

    # 注册MCP工具
    await orchestrator.register_mcp("search", "python", ["search_mcp_server.py"])
    await orchestrator.register_mcp("filesystem", "python", ["filesystem_mcp_server.py"])

    # 注册A2A Agent
    await orchestrator.register_a2a("analyzer", "http://localhost:8081")

    # 执行工作流：搜索信息 → 分析数据
    workflow = [
        {
            "name": "搜索信息",
            "type": "mcp",
            "server": "search",
            "tool": "web_search",
            "args": {"query": "2026 AI Agent 发展趋势"}
        },
        {
            "name": "分析结果",
            "type": "a2a",
            "agent": "analyzer",
            "task": {
                "id": "task-001",
                "skill": "analyze",
                "params": {"data": "搜索结果数据..."}
            }
        }
    ]

    results = await orchestrator.execute_workflow(workflow)
    for r in results:
        print(f"\n➡️ {r['step']}:")
        print(json.dumps(r['result'], ensure_ascii=False, indent=2))

if __name__ == "__main__":
    asyncio.run(main())

五、Docker Compose部署

# docker-compose.yml
version: '3.8'

services:
  # Orchestrator
  orchestrator:
    build: ./orchestrator
    ports:
      - "8000:8000"
    volumes:
      - ./workspace:/data/workspace:ro

  # A2A Agents
  analyzer-agent:
    build: ./agents/analyzer
    ports:
      - "8081:8081"

  writer-agent:
    build: ./agents/writer
    ports:
      - "8082:8082"

  # MCP Servers
  mcp-search:
    build: ./mcp_servers/search
    environment:
      - TAVILY_API_KEY=${TAVILY_API_KEY}

  mcp-filesystem:
    build: ./mcp_servers/filesystem
    volumes:
      - ./shared_data:/data/shared:ro

  mcp-database:
    build: ./mcp_servers/database
    environment:
      - DB_CONNECTION_STRING=${DB_CONNECTION_STRING}

networks:
  default:
    name: agent-mesh

六、生产部署检查清单

七、总结

• MCP总线 + A2A协议 = 完整的Agent互操作性方案
• MCP把所有工具抽象成统一接口，所有Agent都能调用
• A2A让不同框架的Agent能互相通信，解决的是协作层的问题
• Orchestrator是核心，负责任务分发和结果聚合
• Docker Compose一键部署，适合生产环境

💡 建议收藏本文。 整套代码可以直接拿去做多Agent系统的脚手架。

📤 转发给搭建Agent基础设施的同事。 多Agent协作是2026年的主流架构，这套方案可以直接参考。

📚 Agent互操作性系列 (一) MCP与A2A协议对比：两大标准拆解 → (二) 用MCP和A2A打通不同Agent框架（本篇）