MCP+A2A双协议栈实战：从单Agent调用到跨组织协作

想让你的AI Agent既能调用本地工具，又能跨公司协作审批？单靠一个协议搞不定。

MCP管"工具调用"，A2A管"Agent对话"——两个协议搭在一起，才是完整的多Agent协作方案。本文用一个报销审批场景，手把手带你跑通整个流程。

先搞清楚：MCP和A2A到底管什么

很多开发者把这两个协议搞混了。简单说：

关键认知：它们是互补关系，不是替代关系。

一个Agent可以通过MCP接入各种工具（数据库、API、文件系统），多个Agent之间可以通过A2A互相发现、互相委托任务。复杂系统中，两者缺一不可。

第一步：用MCP让Agent调用本地工具

假设你有一个"财务Agent"，需要查询公司数据库里的报销记录。

1. 搭建MCP Server

用Python写一个简单的MCP工具服务，暴露数据库查询能力：

# mcp_finance_server.py
from mcp.server import Server
from mcp.types import Tool, TextContent
import sqlite3

server = Server("finance-tools")

@server.tool()
async def query_expense(employee_id: str, month: str) -> list[TextContent]:
    """查询指定员工某月的报销记录"""
    conn = sqlite3.connect("company_finance.db")
    cursor = conn.execute(
        "SELECT * FROM expenses WHERE employee_id=? AND month=?",
        (employee_id, month)
    )
    rows = cursor.fetchall()
    conn.close()
    return [TextContent(type="text", text=str(rows))]

@server.tool()
async def get_policy(max_amount: float) -> list[TextContent]:
    """根据金额返回审批政策"""
    if max_amount <= 500:
        return [TextContent(type="text", text="部门经理审批即可")]
    elif max_amount <= 5000:
        return [TextContent(type="text", text="需要财务总监审批")]
    else:
        return [TextContent(type="text", text="需要CEO审批")]

if __name__ == "__main__":
    server.run(transport="stdio")

2. Agent端接入MCP

from mcp.client import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def run_finance_agent():
    server_params = StdioServerParameters(
        command="python",
        args=["mcp_finance_server.py"]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # Agent现在可以调用工具了
            result = await session.call_tool(
                "query_expense",
                arguments={"employee_id": "EMP001", "month": "2026-05"}
            )
            print(result)

到这里，你的Agent已经能安全地访问本地数据库了。但问题来了——如果报销金额超过5000块，需要CEO审批，而CEO用的是另一套系统怎么办？

第二步：用A2A实现跨组织Agent协作

这时候A2A协议登场。它让不同组织的Agent可以互相发现、委托任务。

1. 定义Agent能力卡片（Agent Card）

每个A2A兼容的Agent都需要发布一个能力描述：

{
  "name": "CEO-Approval-Agent",
  "description": "处理高额报销审批",
  "url": "https://ceo-agent.company.com/a2a",
  "skills": [
    {
      "name": "approve_high_expense",
      "description": "审批5000元以上报销",
      "inputSchema": {
        "type": "object",
        "properties": {
          "employee_id": {"type": "string"},
          "amount": {"type": "number"},
          "description": {"type": "string"},
          "receipt_url": {"type": "string"}
        }
      }
    }
  ]
}

2. 财务Agent发起跨组织任务委托

import httpx

async def delegate_to_ceo(expense_data: dict):
    """通过A2A协议将审批任务委托给CEO Agent"""
    
    # Step 1: 发现CEO Agent的能力
    async with httpx.AsyncClient() as client:
        card = await client.get("https://ceo-agent.company.com/.well-known/agent.json")
        agent_info = card.json()
    

![配图](https://yitb.com/usr/uploads/covers/cover_mcp_20260705_081451.jpg)

    # Step 2: 发送任务请求
    task_request = {
        "jsonrpc": "2.0",
        "method": "tasks/send",
        "params": {
            "message": {
                "role": "user",
                "parts": [
                    {
                        "type": "text",
                        "text": f"请审批员工{expense_data['employee_id']}的报销申请，金额{expense_data['amount']}元，用途：{expense_data['description']}"
                    },
                    {
                        "type": "file",
                        "file": {"url": expense_data["receipt_url"]}
                    }
                ]
            }
        },
        "id": "task-001"
    }
    
    async with httpx.AsyncClient() as client:
        response = await client.post(
            agent_info["url"],
            json=task_request,
            headers={"Content-Type": "application/json"}
        )
        
    result = response.json()
    
    # Step 3: 处理返回结果（可能是直接回复，也可能是需要人工介入）
    if result["result"]["status"]["state"] == "completed":
        return result["result"]["artifacts"]  # 审批结果
    else:
        return {"status": "pending", "task_id": result["result"]["id"]}

3. CEO Agent端处理审批

# ceo_agent_server.py - 使用A2A SDK搭建
from a2a.server import A2AServer, TaskHandler

class CEOApprovalHandler(TaskHandler):
    async def handle(self, task):
        # 解析任务内容
        expense_info = self.parse_expense(task["message"])
        
        # CEO Agent可以有自己的MCP工具来查更多上下文
        employee_record = await self.mcp_client.call_tool(
            "query_employee_performance",
            {"employee_id": expense_info["employee_id"]}
        )
        
        # 自动化决策或推送给CEO人工审批
        if expense_info["amount"] < 10000 and employee_record["rating"] > 4:
            return self.approve(task, reason="金额合理且员工绩效优秀")
        else:
            return self.escalate(task, notify="ceo@company.com")

server = A2AServer(host="0.0.0.0", port=8080)
server.register_handler(CEOApprovalHandler())
server.run()

双协议栈集成：完整流程串起来

把MCP和A2A组合在一起，报销审批的完整链路是这样的：

员工提交报销 → 财务Agent（MCP查数据库+查政策）
                    ↓ 金额>5000
              财务Agent（A2A委托CEO Agent）
                    ↓
              CEO Agent（MCP查员工绩效）
                    ↓
              CEO Agent（A2A返回审批结果）
                    ↓
              财务Agent通知员工

核心集成代码：

async def process_expense(employee_id, amount, description, receipt_url):
    # Phase 1: MCP - 查询和验证
    records = await mcp_session.call_tool("query_expense", {"employee_id": employee_id})
    policy = await mcp_session.call_tool("get_policy", {"max_amount": amount})
    
    # Phase 2: A2A - 根据政策决定是否跨组织协作
    if "CEO审批" in policy[0].text:
        result = await delegate_to_ceo({
            "employee_id": employee_id,
            "amount": amount,
            "description": description,
            "receipt_url": receipt_url
        })
        return result
    elif "财务总监" in policy[0].text:
        result = await delegate_to_cfo(...)  # 类似逻辑
        return result
    else:
        return {"status": "auto_approved", "reason": "金额在部门经理审批范围内"}

实际商业价值

这套双协议栈方案已经在几个场景落地：

1. 企业内部审批自动化：报销、采购、请假等流程，减少人工流转环节60%以上
2. 跨公司供应链协作：订单Agent与供应商Agent自动对接，对账效率提升3倍
3. SaaS平台Agent生态：通过A2A让第三方Agent接入你的平台，MCP保证工具调用安全

一个中等规模企业部署这套方案，预计每月节省200+小时人工审批时间，按人力成本折算约2-3万元/月。

下一步行动

1. 今天就能做：克隆MCP官方SDK，跑通一个本地工具调用demo
2. 本周目标：搭建你的第一个MCP Server，让Agent能查你自己的数据库
3. 进阶挑战：参考A2A协议规范，写两个Agent互相委托任务的最小案例

协议是骨架，业务是血肉。先跑通最小闭环，再逐步扩展Agent数量和工具复杂度。

有问题？来龙虾官网(yitb.com)社区交流，我们有一堆实战踩坑经验等着你。