MCP A2a Langgraph Pipelining

MCP-A2A-LangGraph-流水线

一个复杂的多代理编排系统，结合了 谷歌ADK, LangGraph, 模型上下文协议（MCP），以及 A2A（代理人对代理人） 通信以构建可扩展的分布式AI代理管道。

🎯 概述

该项目实现了一个分层多代理架构，其中：

• 根编排器（ADK代理）：管理任务委派和代理间沟通
• LangGraph代理：通过工具集成处理代理工作流
• MCP工具：通过模型上下文协议提供数学和天气功能
• A2A服务器：允许通过HTTP/RPC进行代理间通信

架构图

┌─────────────────────────────────────────────────────────────┐
│              Root Orchestrator (ADK Agent)                   │
│  - Model: gemini-2.0-flash                                   │
│  - Tools: [A2A Remote Tool]                                  │
└──────────────────────┬──────────────────────────────────────┘
                       │
                       ├── Sub-Agent 1 (LangGraph Agent)
                       │   ├── Chatbot Node (LLM with tools)
                       │   ├── Tool Node (MCP integration)
                       │   └── State Management
                       │
                       └── A2A Server (Remote Agent Caller)
                           └── HTTP/RPC Communication

📁 项目结构

Pipelines/
├── main.py                          # Application entry point
├── requirements.txt                 # Python dependencies
├── README.md                        # This file
├── agents.py                        # Additional agent utilities
├── tools_server.py                  # MCP tools server setup
│
└── app/
    ├── __init__.py
    ├── adk/
    │   ├── adk_agent.py            # ADK orchestrator builder
    │   ├── adk_client_tool.py       # A2A client tool implementation
    │   └── __pycache__/
    │
    ├── graph/
    │   ├── langgraph_app.py         # LangGraph workflow definition
    │   ├── state.py                 # Agent state schema
    │   ├── tool_server.py           # MCP server configuration
    │   └── __pycache__/
    │
    └── server/
        └── a2a_server.py            # A2A HTTP server implementation

🚀 关键组件

1. ADK代理（app/ADK/ADK_Agent.py）

协调所有子代理并管理工具执行的根编排器。

orchestrator = ADKAgent(
    name="root_orchestrator",
    model="gemini-2.0-flash",
    instruction="Delegate tasks to LangGraph agent. Use A2A tool when needed.",
    tools=[remote_tool],
    sub_agents=[graph_agent]
)

责任：

• 任务编排和委派
• 模型调用（Gemini 2.0 Flash）
• 工具管理和执行
• 子代理生命周期管理

2. LangGraph代理（app/graph/LangGraph_app.py）

通过状态管理和工具集成实现代理工作流。

主要特点：

• 状态图：使用消息历史记录管理对话状态
• 聊天机器人节点：带工具绑定的LLM推理（AWS Bedrock Nova）
• 工具节点：执行MCP工具（数学、天气）
• 条件边：需要时提供工具路线

工作流程：

START → chatbot (LLM inference) → tools_condition → tool_execution → chatbot → END

3. MCP工具集成

连接到MCP服务器以执行工具。

可用工具：

• 数学运算：基本算术计算
• 天气信息：实时天气数据

配置：

client = MultiServerMCPClient({
    "math_weather": {
        "transport": "sse",
        "url": "http://localhost:8000/sse"
    }
})

4. A2A服务器（app/Server/A2A_Server.py）

将代理作为HTTP服务公开，用于远程代理通信。

终点：

• POST /a2a：代理方法调用的RPC端点
• GET /.well-known/agent.json：代理卡/元数据

特征：

• 内存代理运行程序
• JSON-RPC 2.0支持
• FastAPI服务器
• 通过知名端点发现代理

5. A2A客户端工具（app/adk/adk_Client_Tool.py）

使编排器能够通过A2A协议调用远程代理。

remote_tool = A2AClientTool(
    name="call_remote_agent",
    target_agent_url="http://remote-agent.com/a2a"
)

🔧 安装

先决条件

• Python 3.10+
• 谷歌云证书（适用于ADK和Gemini API）
• AWS证书（适用于Bedrock）

设置步骤

1. 克隆仓库

git clone 
cd Pipelines

1. 创建虚拟环境

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

1. 安装依赖项

pip install -r requirements.txt

1. 配置环境变量

# Set up Google Cloud credentials
export GOOGLE_APPLICATION_CREDENTIALS="path/to/credentials.json"

# Set up AWS credentials
export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export AWS_REGION="us-east-1"

📦 依赖项

▶️ 运行应用程序

选项1：启动编排器

python main.py

这将：

• 初始化ADK代理编排器
• 构建LangGraph工作流
• 连接到MCP工具服务器
• 日志初始化状态
• 让代理为请求做好准备

选项2：作为A2A服务器运行

# Terminal 1: Start MCP tools server
python tools_server.py

# Terminal 2: Start A2A server
uvicorn app.server.a2a_server:app --port 9000 --reload

选项3：运行自检

python app/graph/langgraph_app.py

这将使用示例查询测试LangGraph工作流：

"What is (3 + 5) * 12 and weather in Tokyo?"

🔄 工作流示例

示例1：简单计算

# The LLM detects a math query and delegates to MCP math tool
User: "Calculate (3 + 5) * 12"
→ Chatbot detects tool call
→ Tool Node executes via MCP
→ Result: 96

示例2：多工具请求

# Multiple tools are called in sequence
User: "What is (10 + 5) * 2 and weather in Tokyo?"
→ Math tool: (10 + 5) * 2 = 30
→ Weather tool: Returns Tokyo weather
→ Chatbot synthesizes response

示例3：远程代理委派

# Orchestrator delegates to remote agent
User: "Process this complex request"
→ Orchestrator evaluates task
→ If specialized handler needed: call A2A remote agent
→ Remote agent processes and returns result
→ Orchestrator synthesizes final response

📊 状态管理

代理状态模式（app/graph/State.py）

class State(TypedDict):
    messages: Annotated[list, add_messages]

特征：

• 消息历史记录：保持对话上下文
• add_message：用于消息附加的缩减功能
• 类型安全：带TypedDict的完整类型提示

🛠️ 定制

添加新的MCP工具

1. 在MCP服务器中创建一个工具
2. 更新 langgraph_app.py MCP客户端配置：

client = MultiServerMCPClient({
    "your_tools": {
        "transport": "sse",
        "url": "http://localhost:8000/sse"
    }
})

更改LLM模型

编辑 app/graph/langgraph_app.py:

# Option 1: Use different Bedrock model
llm = ChatBedrockConverse(model="amazon.nova-pro-v1:0")

# Option 2: Use Gemini directly in orchestrator
# (Already configured in adk_agent.py)

添加子代理

在 app/adk/adk_agent.py:

sub_agents = [
    graph_agent,
    your_new_agent_1,
    your_new_agent_2,
]

orchestrator = ADKAgent(
    ...
    sub_agents=sub_agents
)

🧪 测试

单元测试

pytest tests/

手动测试

# Test LangGraph workflow
python app/graph/langgraph_app.py

# Test A2A server
curl -X POST http://localhost:9000/a2a \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "message/send",
    "params": {"prompt": "What is 5 + 3?"}
  }'

🔐 安全注意事项

1. API密钥：使用环境变量，从不提交凭据
2. 跨域资源共享：如果从外部源访问，请配置FastAPI CORS
3. 认证：将身份验证中间件添加到A2A服务器端点
4. 速率限制：对公共端点实施速率限制
5. 输入验证：验证所有传入的提示和参数

🐛 故障排除

问题：MCP工具未加载

Solution: Ensure MCP server is running on http://localhost:8000/sse

问题：未找到基岩证书

Solution: Verify AWS credentials are properly configured
export AWS_ACCESS_KEY_ID=...
export AWS_SECRET_ACCESS_KEY=...

问题：远程代理无法访问

Solution: Check A2A server is running and URL is correct
curl http://remote-agent.com/a2a -X OPTIONS

📚 学习资源

• 谷歌ADK文档: https://cloud.google.com/docs/adk
• LangGraph: https://langchain-ai.github.io/langgraph/
• 模型上下文协议: https://modelcontextprotocol.io/
• A2A协议: https://github.com/CloudWeaver-AI/a2a-spec
• AWS基岩: https://aws.amazon.com/bedrock/

🤝 贡献

1. 创建要素分支
2. 进行更改
3. 彻底测试
4. 提交拉取请求

📝 许可证

该项目是Google ADK生态系统的一部分，并遵循其许可条款。

👤 作者

赫曼斯·卡尔蒂克\ 附加培训项目-多代理编排

🎓 学习目标

通过研究这个代码库，你会学到：

• ✅ 多代理架构设计模式
• ✅ LangGraph工作流编排
• ✅ 模型上下文协议集成
• ✅ 代理间通信（A2A）
• ✅ 工具增强LLM代理
• ✅ 异步Python模式
• ✅ 分布式系统中的状态管理
• ✅ FastAPI服务器开发

📞 支持

对于问题和疑问：

• 检查故障排除部分
• 查看应用程序中的示例代码/
• 使用自检脚本进行测试
• 启用调试日志记录以进行详细跟踪