Multi Agent Customer Service System

多代理客户服务系统-A2A和MCP

该项目实现了 多代理客户服务自动化系统 它结合了：

• A2A（代理对代理）通信 –协调专业代理
• MCP（模型上下文协议） –通过工具访问客户和票务数据

它模拟了真实的工作流程，如帐户帮助、计费问题、工单创建和高级客户报告。

______________________________________________________________________

1.项目结构

Applied GenAI_Multi Ag/
├── a2a_servers.py        # Starts A2A HTTP servers for specialist agents
├── agents.py             # Customer Data Agent & Support Agent + Router (Orchestrator)
├── database_setup.py     # Creates SQLite schema & inserts sample data
├── mcp_server.py         # MCP server exposing DB operations as MCP tools
├── main.py               # end-to-end test scenarios
├── requirements.txt      # Python dependencies
└── support.db            # SQLite database file (generated by database_setup.py)

______________________________________________________________________

2.代理角色

路由器代理（编排器）

• 接收客户查询
• 检测意图
• 通过A2A协调专业代理
• 实现任务分配、协商和多步骤工作流
• 向用户返回最终答案

客户数据代理

• 通过以下方式连接到MCP服务器 McpToolset

• 使用MCP工具：

- 检索客户资料 - 列出具有筛选器的客户 - 更新客户字段 - 获取门票历史记录 - 创建门票（需要时）

• 将原始JSON总结为简洁、结构化的信息

支援谍员

• 处理一般支持、计费、订阅和帐户问题
• 通过MCP工具读取上下文（客户信息+工单历史记录）
• 决定何时升级并创建高优先级票证
• 产生富有同理心、面向用户的响应

这两家专业代理都被暴露为A2A服务 a2a_servers.py.

______________________________________________________________________

3.MCP服务器和数据库

MCP服务器在 mcp_server.py.

数据库初始化- database_setup.py

database_setup.py 创建并填充SQLite数据库：

桌子：

客户

id          INTEGER PRIMARY KEY
name        TEXT NOT NULL
email       TEXT
phone       TEXT
status      TEXT ('active' or 'disabled')
created_at  TIMESTAMP
updated_at  TIMESTAMP

门票

id          INTEGER PRIMARY KEY
customer_id INTEGER  -- FK to customers.id
issue       TEXT NOT NULL
status      TEXT ('open', 'in_progress', 'resolved')
priority    TEXT ('low', 'medium', 'high')
created_at  TIMESTAMP

脚本生成数据库文件 support.db 有样品客户和门票。

MCP工具（in mcp_server.py)

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

• get_customer(customer_id)
• list_customers(status, limit)
• update_customer(customer_id, data)
• create_ticket(customer_id, issue, priority)
• get_customer_history(customer_id)

这些工具被包裹在MCP响应中，并通过 服务器发送事件（SSE） 在：

http://127.0.0.1:5000/mcp

健康检查端点也可用：

http://127.0.0.1:5000/health

______________________________________________________________________

4.A2A协调

a2a_servers.py 使用Google ADK的A2A支持来公开：

每个代理都有一个 AgentCard 描述其技能，并运行A2A服务器 uvicorn. 路由器（in main.py)通过A2A客户端与这些URL进行对话，并记录：

[R→Data] ...
[Data→R] ...
[R→Support] ...
[Support→R] ...

此显式日志记录演示了分配的代理间通信。

______________________________________________________________________

5.场景（任务要求）

路由器在 main.py 实施和测试以下内容：

场景1——任务分配

查询示例：

I need help with my account, customer ID 5

______________________________________________________________________

情景2——谈判/升级

查询示例：

I want to cancel my subscription but I'm having billing issues

______________________________________________________________________

场景3——多步协调

查询示例：

What's the status of all high-priority tickets for premium customers?

______________________________________________________________________

6.附加测试场景

main.py 还演示了：

• 简单查询

Get customer information for ID 5

• 升级

I've been charged twice, please refund immediately!

• 多意图

Update my email to new@email.com and show my ticket history. My customer ID is 1.

______________________________________________________________________

7.安装和设置

1.️⃣ 克隆仓库

git clone https://github.com//applied-genai-multi-agent.git
cd "Applied GenAI_Multi Agent"

2.️⃣ 创建并激活虚拟环境

python -m venv venv
# Windows
venv\Scripts\activate
# macOS / Linux
source venv/bin/activate

3.️⃣ 安装依赖项

pip install -r requirements.txt

4.️⃣ 设置环境变量

您必须拥有Google Gemini API密钥。

# Windows (PowerShell)
$env:GOOGLE_API_KEY="YOUR_API_KEY"
# macOS / Linux
export GOOGLE_API_KEY="YOUR_API_KEY"

______________________________________________________________________

8.运行系统（3个进程）

步骤1-初始化数据库

如果数据库不存在或您希望彻底重置：

python database_setup.py

这将创建/覆盖 support.db 使用样本数据。

______________________________________________________________________

步骤2–启动MCP服务器

python mcp_server.py

预期产出（简化）：

✅ Database initialized / using support.db
✅ MCP server running at http://127.0.0.1:5000/mcp

______________________________________________________________________

步骤3–启动A2A代理服务器

在第二个终端中：

python a2a_servers.py

您应该看到：

Customer Data Agent: http://127.0.0.1:10030/.well-known/agent.json
Support Agent      : http://127.0.0.1:10031/.well-known/agent.json

______________________________________________________________________

步骤4–运行路由器和场景

在第三个终端中：

python main.py

您将看到以下日志：

==== New Query ====
USER: Get customer information for ID 5

[R→Data] ...
[Data→R] ...
[R→Support] ...
[Support→R] ...
[Router] Final answer:
...

这些日志可以被捕获并作为以下证据包含在任务中：

• A2A通信
• MCP工具使用
• 多智能体协调

______________________________________________________________________

9.要求

requirements.txt:

google-adk==1.9.0
a2a-sdk==0.3.0
google-generativeai
httpx
uvicorn
starlette
Flask
flask-cors
termcolor
pydantic
typing_extensions

______________________________________________________________________

10.我学到了什么（总结）

您可以将其用于结论部分：

Through this project, I learned how to design and implement
multi-agent systems using A2A communication patterns and
the Model Context Protocol (MCP). I gained hands-on experience
in coordinating specialized agents—Router, Support, and Data
Agent—while ensuring effective tool usage and state transitions
within a LangGraph workflow.

The biggest challenge was preventing infinite loops and
ensuring correct routing based on agent intents. I learned how
to encode agent-to-agent negotiation signals and design robust
routing rules that allow complex, multi-step coordination
(e.g., escalation, multi-intent queries, and chained data fetches).
This project significantly improved my understanding of modern
agent frameworks and production-grade workflow orchestration.