A2A Protocol

概念介绍

A2A（Agent-to-Agent）协议是由谷歌发起、现由 Linux 基金会托管的开放标准，旨在为不同的 AI 智能体提供一种“通用语言”。它通过“Agent Card”让智能体动态发现彼此的能力，并利用 HTTP、JSON-RPC 2.0 和 Server-Sent Events（SSE）等标准技术，协调执行从简单查询到需要实时流式传输的长期任务。其重要性在于打破了智能体之间的生态壁垒，实现了跨平台、跨框架的互操作性。这使得企业可以将不同供应商的智能体组装成协同工作的多智能体系统，从而自动化复杂业务流程，无需为每一个新组合编写定制化的集成代码。—— DeepSeek

为什么使用 A2A 协议？

• 互操作性：连接基于不同平台构建的 Agent，以创建强大的复合型 AI 系统。
• 复杂工作流程：使 Agent 能够分配子任务、交换信息以及协调行动，解决单 Agent 无法处理的复杂问题。
• 安全与不透明：Agent 之间交互无需共享内部内存、工具或专有逻辑，确保安全性并保护知识产权。

A2A 如何与 MCP 协同工作？

示例逻辑

演示任务：

• 1 个 Main：主调用程序
• 1 个 Registry：提供 Agent 地址
• 1 个 Calculator Agent：拆解任务 + 决策调用哪个 Agent
• 2 个 实际计算 Agent：一个执行加法任务，一个执行减法任务

示例运行

示例目录结构：

├── main.py
├── registry.py
├── agent_calculator.py
├── agent_add.py
├── agent_sub.py

开 5 个终端，分别运行：

python registry.py
python agent_add.py
python agent_sub.py
python agent_calculator.py
python main.py

示例代码

main.py 文件：

# main.py
import requests

CALCULATOR_URI = "http://localhost:5003"

# 示例任务
tasks = [
    {"operation": "add", "numbers": [5, 3]},
    {"operation": "sub", "numbers": [5, 3]}
]

for task in tasks:
    print(f"[Main] Sending task: {task}")
    artifact = requests.post(f"{CALCULATOR_URI}/calculate", json=task).json()
    print(f"[Main] Received artifact: {artifact}\n")

registry.py 文件：

# registry.py
from flask import Flask, request, jsonify

app = Flask(__name__)
agents = {}  # 存储 Agent Card

@app.route("/register", methods=["POST"])
def register_agent():
    agent_card = request.json
    agents[agent_card['name']] = agent_card
    print(f"[Registry] Registered agent: {agent_card['name']}")
    return jsonify({"status": "ok"})

@app.route("/discover", methods=["GET"])
def discover_agents():
    print(f"[Registry] Agents discovered: {list(agents.keys())}")
    return jsonify(list(agents.values()))

if __name__ == "__main__":
    print("[Registry] Starting Registry on http://localhost:5000")
    app.run(port=5000)

agent_calculator.py 文件，这是 A2A 的关键：动态发现 + 模拟 LLM 决策 + 链式调用

# calculator.py
from flask import Flask, request, jsonify
import requests

app = Flask(__name__)
REGISTRY_URI = "http://localhost:5000"

def find_agent(op):
    agents = requests.get(f"{REGISTRY_URI}/discover").json()
    for agent in agents:
        if agent['capability'] == op:
            print(f"[Calculator] Found agent {agent['name']} for operation '{op}'")
            return agent['uri']
    return None

@app.route("/calculate", methods=["POST"])
def calculate():
    task = request.json
    op = task['operation']
    agent_uri = find_agent(op)
    if not agent_uri:
        return jsonify({"error": "No agent found"}), 404
    artifact = requests.post(f"{agent_uri}/task", json=task).json()
    print(f"[Calculator] Received artifact: {artifact}")
    return jsonify(artifact)

if __name__ == "__main__":
    print("[Calculator] Running on http://localhost:5003")
    app.run(port=5003)

agent_add.py 文件：

# agent_add.py
from flask import Flask, request, jsonify
import requests

app = Flask(__name__)

# Agent Card
agent_card = {
    "name": "agent_add",
    "capability": "add",
    "uri": "http://localhost:5001"
}

REGISTRY_URI = "http://localhost:5000"

# 注册到 Registry
requests.post(f"{REGISTRY_URI}/register", json=agent_card)

@app.route("/task", methods=["POST"])
def handle_task():
    task = request.json
    print(f"[Add Agent] Received task: {task}")
    a, b = task['numbers']
    artifact = {"result": a + b}  # Artifact
    print(f"[Add Agent] Returning artifact: {artifact}")
    return jsonify(artifact)

if __name__ == "__main__":
    print("[Add Agent] Running on http://localhost:5001")
    app.run(port=5001)

agent_sub.py 文件：

# agent_sub.py
from flask import Flask, request, jsonify
import requests

app = Flask(__name__)

# Agent Card
agent_card = {
    "name": "agent_sub",
    "capability": "sub",
    "uri": "http://localhost:5002"
}

REGISTRY_URI = "http://localhost:5000"

# 注册到 Registry
requests.post(f"{REGISTRY_URI}/register", json=agent_card)

@app.route("/task", methods=["POST"])
def handle_task():
    task = request.json
    print(f"[Sub Agent] Received task: {task}")
    a, b = task['numbers']
    artifact = {"result": a - b}  # Artifact
    print(f"[Sub Agent] Returning artifact: {artifact}")
    return jsonify(artifact)

if __name__ == "__main__":
    print("[Sub Agent] Running on http://localhost:5002")
    app.run(port=5002)

参考

[1] https://a2a-protocol.org/

[2] https://a2acn.com/