Agent2Agent协议介绍-EW帮帮网

什么是Agent2Agent

Agent2Agent（A2A）协议：Google 于 2025 年 4 月 9 日发布的 A2A协议，A2A 是一个开放协议，目的是促进 AI Agent之间的协作，适用于大规模、多智能体的部署。
A2A 基于五个核心原则：

拥抱智能体能力：支持自然、非结构化的协作模式。
利用现有标准：使用 HTTP、Server-Sent Events (SSE) 和 JSON-RPC，
确保与现有系统的兼容性。
默认安全：支持企业级认证和授权，启动时与 OpenAPI 保持一致。
支持长期任务：处理从快速任务到深入研究的任务，提供实时反馈、通知和状态更新。
多模态支持：支持文本、音频、视频流等多模态通信。

在这里插入图片描述
Agent Card：公共元数据文件，位于 /.well-known/agent.json，描述智能体的能力、技能、端点 URL 和认证要求，用于能力发现。

A2A 服务器：暴露 HTTP 端点，实现 A2A 协议方法，管理任务执行。
A2A 客户端：应用程序或智能体，消费 A2A 服务，通过发送 tasks/send 或 tasks/sendSubscribe 请求与服务器交互。

A2A 的典型工作流程

发现：客户端从 /.well-known/agent.json 获取 Agent Card，了解智能体能力
启动：客户端发送任务请求：
使用 tasks/send 处理即时任务，返回最终 Task 对象
使用 tasks/sendSubscribe 处理长期任务，服务器通过 SSE 事件发送更新
处理：服务器处理任务，可能涉及流式更新或直接返回结果
交互（可选）：若任务状态为 input-required，客户端可发送更多消息，使
用相同 Task ID 提供输入
完成：任务达到终端状态（如 completed、failed 或 canceled）
此流程支持简单任务和需要多次交互的复杂任务，特别适合多模态通信环境

A2A 与 MCP的关系

A2A 与MCP 是互补的，可以将A2A看成是一个电话簿，MCP看成是工具说明书

MCP（模型上下文协议）用于工具和资源

通过结构化的输入/输出将代理连接到工具、API 和资源。
Google ADK 支持 MCP 工具，允许与Agent一起使用广泛的 MCP 服务器。

A2A（Agent2Agent 协议）用于Agent之间的协作

在不同Agent之间实现动态的、多模态的通信，而无需共享内存、资源和工具。
这是一个由社区驱动的开放标准。
在 Google ADK、LangGraph 和 Crew.AI 中有提供的参考示例

MCP 确保Agent能访问数据和工具（如通过 Google Drive、Slack、GitHub 等）。
A2A 则让Agent能协作处理这些数据，完成任务。

协议的交集：

Google建议应用将A2A智能体通过AgentCard描述注册为MCP资源。这样，框架既能通过MCP调用工具，又能通过A2A与用户、远程智能体通信，实现无缝协作。
在这里插入图片描述

例如，一个智能体可能通过 MCP 从数据库检索数据，然后通过 A2A 与另一个智能体协作分析数据。

MCP调用：

智能体A的AgentCard声明了“数据库查询”能力，并在MCP注册。
框架通过MCP直接调用智能体A的“数据库查询”资源，获取数据。
此时MCP = “能力执行器”。

A2A通信：
智能体A需要进一步分析数据，但自身不具备该能力。
它通过A2A协议（如发送一条结构化消息）联系智能体B，请求协作。
智能体B的AgentCard声明了“数据分析”能力，并通过A2A返回结果。
此时A2A像是一个“协作总线”。

为什么这样设计

解耦能力与通信：
MCP专注标准化能力调用（类似微服务），A2A专注动态协作（类似聊天）。
避免智能体既要处理功能逻辑，又要处理通信协议。

无缝扩展性：
新智能体只需注册AgentCard到MCP，即可被其他智能体发现和调用。
A2A协议允许智能体动态组队（例如临时组建一个“数据分析小组”）。

兼容现有架构：
MCP可集成传统工具（如数据库、API），A2A可对接人类用户或其他异构智能体。

MCP像“应用商店”：提供标准化服务（如地图、支付）。
A2A像“微信群聊”：智能体在群里用通用语言（协议）讨论如何分工。
AgentCard像“个人资料”：写明“我会修图”“我会翻译”，方便群里@你干活。

多智能体协作demo

TO DO：实现了一个自动根据天气情况决定是否安排篮球活动的场景。
WeatherAgent: 提供天气预报服务的智能体
BasketBallAgent: 负责篮球活动安排的智能体

整体流程如下：

BasketBallAgent接收到用户提供的活动日期
BasketBallAgent通过A2A协议向WeatherAgent请求该日期的天气信息
WeatherAgent返回天气数据（温度和天气状况）
BasketBallAgent根据天气状况做出决策：
如果天气不含"雨"或"雪"，确认安排篮球活动；否则，取消活动并提供取消原因

关键步骤：

A2A协议实现
基于HTTP REST API实现，关键组成部分包括：
• Agent Card: 通过/.well-known/agent.json端点暴露的智能体能力描述
• 标准化任务交互：使用统一的任务提交和结果获取格式
• 身份验证：通过API Key进行授权

WeatherAgent实现
WeatherAgent是一个基于FastAPI实现的服务：

from fastapi import FastAPI, HTTPException
from datetime import date
from pydantic import BaseModel
import uvicorn

app = FastAPI()

# Agent Card声明（通过/.well-known/agent.json暴露）
WEATHER_AGENT_CARD = {
    "name": "WeatherAgent",
    "version": "1.0",
    "description": "提供指定日期的天气数据查询",
    "endpoints": {
        "task_submit": "/api/tasks/weather",
        "sse_subscribe": "/api/tasks/updates"
    },
    "input_schema": {
        "type": "object",
        "properties": {
            "date": {"type": "string", "format": "date"},
            "location": {"type": "string", "enum": ["北京"]}
        },
        "required": ["date"]
    },
    "authentication": {"methods": ["API_Key"]}
}

# 任务请求模型
class WeatherTaskRequest(BaseModel):
    task_id: str
    params: dict

# 模拟天气数据存储
weather_db = {
    "2025-05-08": {"temperature": "25℃", "condition": "雷阵雨"},
    "2025-05-09": {"temperature": "18℃", "condition": "小雨转晴"},
    "2025-05-10": {"temperature": "22℃", "condition": "多云转晴"}
}

@app.get("/.well-known/agent.json")
async def get_agent_card():
    return WEATHER_AGENT_CARD

@app.post("/api/tasks/weather")
async def handle_weather_task(request: WeatherTaskRequest):
    """处理天气查询任务"""
    target_date = request.params.get("date")
    # 参数验证
    if not target_date or target_date not in weather_db:
        raise HTTPException(status_code=400, detail="无效日期参数")
    return {
        "task_id": request.task_id,
        "status": "completed",
        "artifact": {
            "date": target_date,
            "weather": weather_db[target_date]
        }
    }

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

BasketBallAgent实现
BasketBallAgent是调用方，负责消费WeatherAgent的服务：

import requests
import uuid

class BasketBallAgent:
    def __init__(self):
        self.weather_agent_url = "http://localhost:8000"
        self.api_key = "SECRET_KEY"  # 实际应通过安全方式存储

    def _create_task(self, target_date: str) -> dict:
        """创建A2A标准任务对象"""
        return {
            "task_id": str(uuid.uuid4()),
            "params": {
                "date": target_date,
                "location": "北京"
            }
        }

    def check_weather(self, target_date: str) -> dict:
        """通过A2A协议查询天气"""
        # 获取天气智能体能力描述
        agent_card = requests.get(
            f"{self.weather_agent_url}/.well-known/agent.json"
        ).json()
        # 构造任务请求
        task = self._create_task(target_date)
        # 发送任务请求
        response = requests.post(
            f"{self.weather_agent_url}{agent_card['endpoints']['task_submit']}",
            json=task,
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        if response.status_code == 200:
            return response.json()["artifact"]
        else:
            raise Exception(f"天气查询失败: {response.text}")

    def schedule_meeting(self, date: str):
        """综合决策逻辑"""
        try:
            result = self.check_weather(date)
            # print('result=', result)  
            if "雨" not in result["weather"]["condition"] and "雪" not in result["weather"]["condition"]:
                return {"status": "confirmed", "weather": result["weather"]}
            else:
                return {"status": "cancelled", "reason": "恶劣天气"}
        except Exception as e:
            return {"status": "error", "detail": str(e)}

# 使用示例
if __name__ == "__main__":
    meeting_agent = BasketBallAgent()
    result = meeting_agent.schedule_meeting("2025-05-08")
    # result = meeting_agent.schedule_meeting("2025-05-10")
    print("篮球安排结果:", result)

Agent2Agent协议介绍

什么是Agent2Agent

A2A 的典型工作流程

A2A 与 MCP的关系

多智能体协作demo

网站公告

今日签到

热门文章

最新发布