MCP和A2A实战教程：构建智能客服系统

MCP和A2A实战教程：构建智能客服系统

在这个教程中，我们将构建一个简单的智能客服系统，展示MCP和A2A如何协同工作。这个系统将包含以下组件：

• 主客服AI（使用A2A与其他AI通信）
• 订单查询AI（使用MCP访问订单数据库）
• 产品推荐AI（使用MCP访问产品目录）
• 支付处理AI（使用MCP处理支付）

1. 环境准备

首先，我们需要安装必要的依赖：

pip install a2a-sdk
pip install fastapi
pip install uvicorn
pip install pydantic

2. 创建基础结构

让我们创建项目的基本结构：

smart_customer_service/
├── agents/
│   ├── __init__.py
│   ├── main_agent.py      # 主客服AI
│   ├── order_agent.py     # 订单查询AI
│   ├── product_agent.py   # 产品推荐AI
│   └── payment_agent.py   # 支付处理AI
├── tools/
│   ├── __init__.py
│   ├── order_tool.py      # 订单数据库工具
│   ├── product_tool.py    # 产品目录工具
│   └── payment_tool.py    # 支付处理工具
└── main.py

3. 实现工具（MCP层）

首先，让我们实现订单查询工具：

# tools/order_tool.py
from pydantic import BaseModel
from typing import Optionalclass OrderQuery(BaseModel):order_id: strcustomer_id: Optional[str] = Noneclass OrderTool:def __init__(self):# 模拟数据库self.orders = {"ORD001": {"status": "已发货","items": ["产品A", "产品B"],"total": 299.99}}def query_order(self, query: OrderQuery) -> dict:if query.order_id in self.orders:return self.orders[query.order_id]return {"error": "订单未找到"}

4. 实现订单查询AI（A2A层）

# agents/order_agent.py
from a2a import Agent, AgentCard, AgentSkill
from tools.order_tool import OrderTool, OrderQueryclass OrderAgent(Agent):def __init__(self):self.order_tool = OrderTool()# 定义Agent Cardself.card = AgentCard(name="订单查询助手",description="帮助查询订单状态和详情",skills=[AgentSkill(id="order-query",name="订单查询",description="查询订单状态和详情",input_modes=["application/json"],output_modes=["application/json"])])async def handle_message(self, message):if message.role == "user":# 解析用户查询query = OrderQuery(**message.content)# 使用MCP调用工具result = self.order_tool.query_order(query)return {"role": "agent","content": result}

5. 实现主客服AI

# agents/main_agent.py
from a2a import Agent, AgentCard, AgentSkill
from typing import Dict, Listclass MainAgent(Agent):def __init__(self):self.connected_agents: Dict[str, Agent] = {}# 定义Agent Cardself.card = AgentCard(name="智能客服助手",description="处理客户咨询和请求",skills=[AgentSkill(id="customer-service",name="客户服务",description="处理客户咨询和请求",input_modes=["text/plain"],output_modes=["text/plain"])])async def connect_to_agent(self, agent_id: str, agent: Agent):self.connected_agents[agent_id] = agentasync def handle_message(self, message):if message.role == "user":# 分析用户意图intent = self.analyze_intent(message.content)if intent == "order_query":# 使用A2A与订单查询AI通信order_agent = self.connected_agents["order_agent"]response = await order_agent.handle_message(message)return response# 处理其他意图...def analyze_intent(self, text: str) -> str:# 简单的意图分析if "订单" in text or "查询" in text:return "order_query"return "general_query"

6. 启动服务

# main.py
import uvicorn
from fastapi import FastAPI
from agents.main_agent import MainAgent
from agents.order_agent import OrderAgentapp = FastAPI()# 创建AI实例
main_agent = MainAgent()
order_agent = OrderAgent()# 连接AI
main_agent.connect_to_agent("order_agent", order_agent)@app.post("/chat")
async def chat(message: str):response = await main_agent.handle_message({"role": "user","content": message})return responseif __name__ == "__main__":uvicorn.run(app, host="0.0.0.0", port=8000)

7. 测试系统

启动服务后，我们可以使用以下命令测试系统：

curl -X POST "http://localhost:8000/chat" \-H "Content-Type: application/json" \-d '{"message": "我想查询订单ORD001的状态"}'

8. 系统工作流程

参考资料

• A2A Protocol Documentation
• Model Context Protocol
• FastAPI Documentation