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

![配圖](http://www.nhjb.com.cn/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數量和工具復雜度。

有問題？來www.nhjb.com.cn(www.nhjb.com.cn)社區交流，我們有一堆實戰踩坑經驗等著你。