Multiagent Customer A2A MCP

基于A2A和MCP的多Agent客户服务系统

一个多代理客户服务系统，演示了代理到代理（A2A）通信和模型上下文协议（MCP）集成。该系统由专门的代理组成，这些代理协调处理客户查询、管理客户数据和处理支持票。

概述

该项目实施了一个三代理系统：

• 路由器代理（编排器）：接收查询、分析意图，并将其发送给专业代理
• 客户数据代理：通过MCP管理客户数据库操作
• 支援谍员：处理客户支持查询和工单管理

建筑

• A2A通信：代理使用代理到代理协议进行通信
• MCP集成：代理通过模型上下文协议服务器访问客户数据
• 数据库：存储客户和支持票的SQLite数据库

先决条件

• Python 3.10或更高版本
• 使用API访问的谷歌云项目
• Google API密钥（适用于Gemini型号）
• （可选）用于访问公共MCP服务器的ngrok帐户

安装

1.克隆存储库

git clone 
cd assignment5

2.创建虚拟环境

python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

3.安装依赖项

pip install --upgrade google-genai google-adk==1.9.0 a2a-sdk==0.3.0 python-dotenv aiohttp uvicorn requests mermaid-python nest-asyncio flask flask-cors termcolor pyngrok

或者从需求文件安装：

pip install -r requirements.txt

4.设置环境变量

创建一个 .env 项目根目录中的文件：

GOOGLE_API_KEY=your_google_api_key_here
GOOGLE_GENAI_USE_VERTEXAI=FALSE
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1

对于谷歌Colab：

• 使用“秘密”功能(🔑 左侧边栏中的图标）来存储：

- GOOGLE_API_KEY：您的Google API密钥 - NGROK_AUTHTOKEN：您的ngrok authtoken（可选，用于公共MCP服务器）

5.初始化数据库

运行数据库安装脚本：

python database_setup.py

这将：

• 创造 support.db SQLite数据库
• 创建 customers 和 tickets 表格
• 可选择插入示例数据
• 运行示例查询以验证设置

按照提示操作：

• 插入示例数据（建议：“y”）
• 运行示例查询（建议：“y”）

用法

选项1:Google Colab（建议用于测试）

步骤1：启动MCP服务器

1. 打开 mcp_integration.ipynb 在谷歌Colab
2. 按顺序运行所有单元格：

- 单元1：安装软件包 - 单元2：数据库设置（创建表和示例数据） - 单元3：定义MCP工具功能 - 单元4：构建MCP服务器 - 单元5：启动MCP服务器

1. 注意MCP服务器URL：

- 当地： http://127.0.0.1:5000/mcp - 吸烟： https://your-ngrok-url.ngrok-free.dev/mcp （如果使用ngrok）

第二步：启动A2A代理

1. 打开 a2a.ipynb 在谷歌Colab
2. 更新 MCP_SERVER_URL 在笔记本中使用步骤1中的MCP服务器URL
3. 按顺序运行所有单元格：

- 单元1：安装软件包 - 单元2：兼容性解决方法 - 单元3：导入库 - 单元4：环境配置 - 单元5：身份验证（仅限Colab） - 单元6：设置日志记录 - 单元格7-10：创建客户数据代理 - 单元格11-14：创建支持代理 - 单元15-16：创建路由器代理 - 单元格17：定义服务器创建功能 - Cell 18：启动所有A2A服务器 - 单元19：验证代理卡 - 单元格20：定义A2A简单客户端 - 单元格21：定义ask_agent辅助函数

1. 等待服务器启动（检查“”的输出✅ 所有代理服务器都已启动！")

步骤3：测试系统

使用 ask_agent() 用于测试查询的函数：

# Simple query
await ask_agent("Get customer information for ID 5")

# Coordinated query
await ask_agent("I'm customer 5 and need help upgrading my account")

# Complex query
await ask_agent("Show me all active customers who have open tickets")

选项2：本地Python环境

步骤1：启动MCP服务器

选项A：使用Jupyter笔记本

jupyter notebook mcp_integration.ipynb

运行所有单元以启动端口5000上的MCP服务器。

选项B：提取Flask应用程序（高级） 从笔记本中提取Flask应用程序并运行：

python mcp_server.py

第二步：启动A2A代理

jupyter notebook a2a.ipynb

1. 更新 MCP_SERVER_URL 到 http://localhost:5000/mcp
2. 运行所有单元格以开始：

- 客户数据代理（端口10021） - 支持代理（端口10020） - 路由器代理（端口10022）

步骤3：测试系统

使用 ask_agent() 辅助功能或 A2ASimpleClient 发送查询。

项目结构

assignment5/
├── README.md                 # This file
├── requirements.txt          # Python dependencies
├── database_setup.py        # Database initialization script
├── mcp_integration.ipynb    # MCP server implementation
├── a2a.ipynb                # A2A agent system implementation
├── support.db               # SQLite database (created after setup)
└── .env                     # Environment variables (create this)

MCP工具

MCP服务器公开了以下工具：

1. get_customer（customer_id）：按ID检索客户
2. list_customers（状态、限制）：列出具有可选状态筛选器的客户
3. update_customer（customer_id，数据）：更新客户信息（数据是一个包含姓名、电子邮件、电话、状态的字典）
4. create_ticket（客户id、问题、优先级）：创建支持票
5. get_customer_history（customer_id）：为客户获取所有门票

配置

端口分配

• MCP服务器：端口5000（默认）
• 客户数据代理：端口10021
• 支援谍员：端口10020
• 路由器代理：端口10022

重要：确保这些端口在以下方面保持一致：

• AgentCard URL定义
• RemoteA2aAgent 端点
• 服务器启动配置

MCP服务器URL

更新 MCP_SERVER_URL A2A笔记本中的变量：

• 当地： http://localhost:5000/mcp
• 吸烟： https://your-ngrok-url.ngrok-free.dev/mcp

测试

测试场景

1. 简单查询：“获取ID 5的客户信息”

- 单代理，直接的MCP呼叫

1. 协同查询：“我是客户5，需要帮助升级我的帐户”

- 多个代理协调：数据获取+支持响应

1. 复杂查询：“显示所有有未结门票的活跃客户”

- 需要数据和支持代理之间进行协商

1. 升级：“我已经被收费两次了，请立即退款！”

- 路由器识别紧急情况并适当路由

1. 多意图：“更新我的电子邮件至new@email.com并显示我的门票历史记录”

- 并行任务执行和协调

故障排除

常见问题

1.端口已在使用中

# Find and kill process using the port (macOS/Linux)
lsof -ti:5000 | xargs kill -9  # For port 5000
lsof -ti:10020 | xargs kill -9  # For Support Agent
lsof -ti:10021 | xargs kill -9  # For Customer Data Agent
lsof -ti:10022 | xargs kill -9  # For Router Agent

# Windows
netstat -ano | findstr :5000
taskkill /PID 
 /F

2.数据库连接错误

• 确保 support.db 存在（运行 database_setup.py)
• 检查 DB_PATH 在MCP集成笔记本中正确设置
• 验证数据库文件是否在正确的位置

3.MCP工具执行错误

• 验证函数签名是否与MCP工具定义完全匹配
• 检查 update_customer 用途 data 参数（dict），而不是单个参数
• 确保 list_customers 包含 limit 参数
• 验证 DB_PATH 在函数单元格中定义

4.代理响应错误(NoneType 没有属性“parts”）

• 这 ask_agent() 该功能包括对“无”内容的防御检查
• 如果错误仍然存在，请检查代理协调和MCP服务器连接
• 验证所有代理是否正在运行且可访问

5.404代理端点上的错误

• 验证所有配置中的端口号是否匹配：

- AgentCard 网址 - RemoteA2aAgent 端点 - 服务器启动端口

• 检查服务器是否已成功启动
• 服务器启动后等待几秒钟再进行测试
• 使用健康检查终结点验证服务器状态

6.Google API关键问题

• 验证 GOOGLE_API_KEY 在中设置正确 .env 或Colab秘密
• 检查API密钥是否具有适用于Gemini模型的适当权限
• 如果出现以下情况，请确保您没有使用Vertex AI GOOGLE_GENAI_USE_VERTEXAI=FALSE
• 使用简单的Google GenAI调用测试API密钥

7.兼容性问题

• 确保您使用的是准确的版本： google-adk==1.9.0 和 a2a-sdk==0.3.0
• 导入ADK模块之前，必须运行笔记本中的兼容性解决方法
• 如果问题仍然存在，请尝试重新启动运行时/内核

调试提示

1. 单独测试试剂：在测试协调之前测试每个代理
2. 检查日志：启用日志记录以查看代理之间的通信
3. 验证MCP服务器：直接使用MCP检查器测试MCP工具：

   npx @modelcontextprotocol/inspector

然后输入您的MCP服务器URL

1. 验证配置：仔细检查所有端口分配和URL
2. 检查事件循环：确保 nest_asyncio.apply() 在笔记本中运行异步代码之前调用

依赖项

核心包

• google-genai：谷歌的生成式人工智能SDK
• google-adk==1.9.0：谷歌代理开发工具包
• a2a-sdk==0.3.0：代理到代理SDK

服务器和协议

• flask：MCP服务器web框架
• flask-cors：MCP服务器的CORS支持
• uvicorn：A2A代理的ASGI服务器
• aiohttp：异步HTTP客户端

公用事业

• python-dotenv：环境变量管理
• nest-asyncio：笔记本电脑的嵌套事件循环支持
• pyngrok：MCP服务器的公共隧道（可选）
• termcolor：彩色端子输出

已知问题和解决方法

兼容性解决方案

对于 google-adk==1.9.0 和 a2a-sdk==0.3.0 兼容性方面，A2A笔记本电脑中包含了一种解决方法。这修补了 A2ACardResolver 引用问题，当googleadk发布1.9.0以上的版本时，应将其删除。

解决方法代码：

import sys
from a2a.client import client as real_client_module
from a2a.client.card_resolver import A2ACardResolver

class PatchedClientModule:
    def __init__(self, real_module) -> None:
        for attr in dir(real_module):
            if not attr.startswith('_'):
                setattr(self, attr, getattr(real_module, attr))
        self.A2ACardResolver = A2ACardResolver

patched_module = PatchedClientModule(real_client_module)
sys.modules['a2a.client.client'] = patched_module

数据库模式

客户表

• id：整数主键
• name：文本不能为空
• email：TEXT
• phone：TEXT
• status：TEXT NOT NULL默认值“活动”（值：“活动”、“禁用”）
• created_at：时间戳
• updated_at：时间戳

门票桌

• id：整数主键
• customer_id：整数不为空（FK表示客户.id）
• issue：文本不能为空
• status：TEXT NOT NULL默认值“open”（值：“open”、“in_progress”、“resolved”）
• priority：TEXT NOT NULL默认值“中等”（值：“低”、“中等”、“高”）
• created_at：日期时间

作者

古凯蒂

致谢

• 谷歌ADK和A2A SDK团队
• 模型上下文协议规范