【全网首发】Python+AI智能体通信：用A2A协议让AI像人类团队一样协作（附完整代码）

还在为AI智能体之间无法互通发愁？这个Python库让不同框架的AI直接对话，代码量减少90%！

写在前面

上周有个做智能客服系统的朋友跟我吐槽：他们用LangChain做了一个问答机器人，用CrewAI做了一个数据分析助手，想让这两个AI协作完成任务——结果光是写通信代码就花了三天时间，最后还是跑不通。

有没有办法让不同框架的AI智能体像人类团队一样自然协作？

答案是肯定的！今天要介绍的A2A（Agent-to-Agent）协议，就是专门解决这个痛点的开放标准。它由Google发起、现由Linux基金会托管，被誉为"智能体界的互联网协议"。最重要的是，Python开发者可以用几行代码就让自己的AI智能体具备跨平台通信能力！

一、A2A是什么？

1.1 A2A的定义

A2A（Agent-to-Agent Protocol）是一个开放标准，让不同厂商、不同框架构建的AI智能体能够直接对话、协作完成任务 。

你可以把它理解为AI智能体界的"通用语"或"互联网"——就像HTTP协议让全球的浏览器和服务器能够通信一样，A2A让全球的AI智能体能够互相"认识"和"协作" 。

1.2 A2A的核心组件

A2A协议定义了几个关键概念，我们可以把它们想象成智能体协作的"工具箱"：

核心组件	一句话解释	主要功能
Agent Card（智能体名片）	智能体的"数字身份证"	描述智能体的名称、能力、访问地址和认证方式
Task（任务）	一次完整的协作单元	带有唯一ID和生命周期状态的工作单元
Message（消息）	智能体间的"对话内容"	携带文本、文件或结构化数据的通信单元
Artifact（产物）	任务的"交付成果"	任务完成后生成的文档、图片等输出

1.3 Agent Card长什么样？

来看一个真实的Agent Card示例（JSON格式）：

{
  "name": "咖啡店助手",
  "description": "希希咖啡店智能服务，可以帮助点咖啡和查询订单",
  "url": "https://agent.example.com/agent",
  "version": "1.0.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "skills": [
    {
      "id": "get_products",
      "name": "获取商品列表",
      "description": "获取当前可点的咖啡饮品列表",
      "examples": ["有什么咖啡推荐", "看看菜单"]
    },
    {
      "id": "create_order",
      "name": "创建订单",
      "description": "帮用户下单点咖啡",
      "examples": ["我要一杯拿铁", "下单两杯美式"]
    }
  ],
  "authentication": {
    "schemes": ["apiKey"]
  }
}

其他智能体可以通过 /.well-known/agent-card.json 路径获取这张"名片"，然后立即了解：这个智能体是做什么的？有哪些技能？如何联系它？

1.4 A2A的应用场景

• 跨平台智能体协作：让LangChain的智能体与CrewAI的智能体直接对话
• 企业级AI系统集成：不同部门、不同供应商的AI系统无缝对接
• 智能客服升级：通用问答机器人调用专业领域的AI助手
• 分布式智能系统：智能体可以部署在不同的服务器、甚至不同的编程语言环境中

二、为什么要用A2A？

2.1 解决的核心痛点

在没有标准化协议之前，连接不同框架的智能体需要大量的定制开发，形成 N × M 集成噩梦：

• 5个智能体 × 10个工具 = 50个定制集成
• 10个智能体 × 20个工具 = 200个定制集成
• 每个集成都需要专门的开发和维护

A2A将这种模式转变为 N + M，就像USB-C统一了设备充电接口，让任何智能体都能与任何其他智能体通信 。

2.2 行业需求

根据2026年3月的最新动态：

• Linux基金会旗下的Agentic AI Foundation已经汇聚了MCP、A2A等主流协议
• IBM的ACP协议已正式并入A2A，两者的优势得以整合
• 中国移动发布了智能体互联网开放网络协议框架
• 华为启动A2A-T协议配套软件的开源计划
• Oracle Fusion Apps在最新版本中集成了A2A支持

这些密集的产业动作传递出一个清晰的信号：智能体通信协议的标准化，正在从"技术探索"走向"产业共识" 。

2.3 A2A vs MCP：不是对手，是队友

很多人容易混淆A2A和另一个热门协议MCP（模型上下文协议）。其实两者是互补协同的关系：

维度	MCP	A2A
核心目的	智能体 ↔ 工具	智能体 ↔ 智能体
类比	AI界的"USB-C"	AI界的"互联网协议"
解决什么问题	如何调用API、查询数据库	如何发现对方、协商任务、传递结果
典型场景	智能体需要获取天气数据	两个智能体协作完成旅行规划

简单来说：MCP给智能体装上了"手脚"去拿工具，A2A让智能体学会了"对话"来协作 。

三、怎么用：Python环境配置与入门

3.1 环境要求

• Python 3.10 或更高版本
• 8GB以上内存（推荐16GB，如果运行本地LLM）

3.2 安装A2A Python SDK

使用pip安装官方SDK：

# 安装核心SDK
pip install a2a-sdk

# 或者安装包含所有扩展功能的版本
pip install "a2a-sdk[all]"

# 如果需要HTTP服务器支持
pip install "a2a-sdk[http-server]"

# 如果需要数据库支持
pip install "a2a-sdk[sql]"  # 所有SQL数据库
# 或单独安装
pip install "a2a-sdk[postgresql]"  # PostgreSQL支持
pip install "a2a-sdk[sqlite]"      # SQLite支持

3.3 快速入门：构建第一个A2A智能体

案例1：创建一个简单的"天气查询"智能体

场景：假设你有一个可以查询天气的AI智能体，现在想把它包装成A2A服务器，让其他智能体可以调用它。

完整代码：

# weather_agent_server.py
"""
A2A天气智能体服务器示例
功能：提供城市天气查询服务
遵循A2A协议规范
"""

import asyncio
from a2a_sdk import A2AServer, AgentCard, AgentSkill, Message, TextPart
from a2a_sdk.transport.http import HTTPServerTransport

# 模拟天气查询函数（实际中可调用真实API）
def get_weather(city: str) -> str:
    """
    模拟获取城市天气
    实际应用中可替换为真实的天气API调用
    """
    weather_data = {
        "北京": "晴朗，25°C，空气质量良好",
        "上海": "多云，28°C，湿度较大",
        "广州": "阵雨，32°C，注意携带雨具",
        "深圳": "阴天，30°C，微风",
        "成都": "阴天，26°C，适宜户外活动"
    }
    return weather_data.get(city, f"抱歉，暂不支持{city}的天气查询")

# 定义智能体处理逻辑
async def handle_message(message: Message) -> Message:
    """
    处理接收到的消息并返回响应
    这是智能体的核心业务逻辑
    """
    # 提取用户消息内容
    user_text = ""
    for part in message.parts:
        if part.type == "text":
            user_text = part.text
            break
    
    # 简单的意图识别（实际中可用LLM）
    if "天气" in user_text:
        # 提取城市名称（简化版）
        import re
        cities = ["北京", "上海", "广州", "深圳", "成都"]
        mentioned_city = None
        for city in cities:
            if city in user_text:
                mentioned_city = city
                break
        
        if mentioned_city:
            weather_info = get_weather(mentioned_city)
            response_text = f"{mentioned_city}的天气：{weather_info}"
        else:
            response_text = "请告诉我您想查询哪个城市的天气？"
    else:
        response_text = "我是天气助手，可以帮您查询城市天气。例如：'北京天气怎么样？'"
    
    # 构建响应消息
    response = Message(
        role="assistant",
        parts=[TextPart(text=response_text)]
    )
    return response

# 创建Agent Card（智能体名片）
agent_card = AgentCard(
    name="WeatherAgent",
    description="天气查询智能体，支持国内主要城市的实时天气信息",
    version="1.0.0",
    url="http://localhost:8000",
    capabilities={
        "streaming": False,
        "pushNotifications": False
    },
    skills=[
        AgentSkill(
            id="query_weather",
            name="查询天气",
            description="获取指定城市的天气信息",
            examples=["北京天气怎么样？", "上海今天会下雨吗？", "广州温度"]
        )
    ],
    authentication={
        "schemes": []  # 本示例暂不需要认证
    }
)

# 创建并启动A2A服务器
async def main():
    # 配置HTTP传输层
    transport = HTTPServerTransport(
        host="0.0.0.0",
        port=8000
    )
    
    # 创建A2A服务器实例
    server = A2AServer(
        agent_card=agent_card,
        message_handler=handle_message,
        transport=transport
    )
    
    print("🚀 天气智能体A2A服务器启动中...")
    print(f"📋 Agent Card地址: http://localhost:8000/.well-known/agent-card.json")
    print(f"💬 智能体端点: http://localhost:8000/")
    print("按 Ctrl+C 停止服务器")
    
    # 启动服务器
    await server.start()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n👋 服务器已停止")

案例2：创建一个客户端调用天气智能体

场景：现在你有一个主控智能体，需要调用刚才创建的天气智能体来获取信息。

完整代码：

# weather_agent_client.py
"""
A2A天气智能体客户端示例
功能：调用远程天气智能体并获取结果
"""

import asyncio
import requests
from a2a_sdk import A2AClient, Message, TextPart
from a2a_sdk.transport.http import HTTPClientTransport

async def discover_agent_card(agent_url: str):
    """
    发现远程智能体的Agent Card
    这是A2A协议的第一步：发现
    """
    card_url = f"{agent_url.rstrip('/')}/.well-known/agent-card.json"
    try:
        response = requests.get(card_url)
        if response.status_code == 200:
            return response.json()
        else:
            print(f"❌ 获取Agent Card失败: HTTP {response.status_code}")
            return None
    except Exception as e:
        print(f"❌ 连接失败: {e}")
        return None

async def query_weather(client: A2AClient, city: str):
    """
    向天气智能体查询天气
    """
    # 创建用户消息
    message = Message(
        role="user",
        parts=[TextPart(text=f"{city}天气怎么样？")]
    )
    
    print(f"📤 发送查询: {city}")
    
    # 发送消息并等待响应
    try:
        response = await client.send_message(message)
        
        # 提取响应文本
        for part in response.parts:
            if part.type == "text":
                print(f"📥 智能体回复: {part.text}")
                return part.text
        
        print("⚠️ 响应中没有文本内容")
        return None
    except Exception as e:
        print(f"❌ 通信失败: {e}")
        return None

async def main():
    """
    主函数：演示与天气智能体的交互
    """
    # 目标智能体地址（假设运行在本地）
    agent_url = "http://localhost:8000"
    
    print("🔍 正在发现天气智能体...")
    
    # 步骤1：发现智能体（获取Agent Card）
    card_info = await discover_agent_card(agent_url)
    if card_info:
        print(f"✅ 发现智能体: {card_info.get('name')}")
        print(f"📝 描述: {card_info.get('description')}")
        print(f"🎯 技能: {[s.get('name') for s in card_info.get('skills', [])]}")
    else:
        print("❌ 无法发现智能体，请确保服务器已启动")
        return
    
    # 步骤2：创建客户端连接
    transport = HTTPClientTransport(base_url=agent_url)
    client = A2AClient(transport=transport)
    
    print("\n" + "="*50)
    print("开始与智能体对话（输入 'quit' 退出）")
    print("="*50)
    
    # 步骤3：交互式对话
    while True:
        city = input("\n🏙️ 请输入城市名称: ").strip()
        if city.lower() in ['quit', 'exit', 'q']:
            break
        
        if city:
            await query_weather(client, city)
        else:
            print("请输入城市名称")
    
    await client.close()

if __name__ == "__main__":
    asyncio.run(main())

运行效果：

🔍 正在发现天气智能体...
✅ 发现智能体: WeatherAgent
📝 描述: 天气查询智能体，支持国内主要城市的实时天气信息
🎯 技能: ['查询天气']

==================================================
开始与智能体对话（输入 'quit' 退出）
==================================================

🏙️ 请输入城市名称: 北京
📤 发送查询: 北京
📥 智能体回复: 北京的天气：晴朗，25°C，空气质量良好

🏙️ 请输入城市名称: 上海
📤 发送查询: 上海
📥 智能体回复: 上海的天气：多云，28°C，湿度较大

四、怎么用好：进阶技巧与最佳实践

4.1 使用AG2框架快速构建A2A智能体

AG2（原AutoGen）框架对A2A提供了原生支持，可以用更少的代码实现同样的功能 ：

# server.py - 使用AG2创建A2A服务器
from ag2 import A2AServer, ConversableAgent

# 创建普通的AG2智能体
weather_agent = ConversableAgent(
    name="weather_agent",
    system_message="你是一个天气助手，可以提供城市天气信息。"
)

# 一行代码包装成A2A服务器
server = A2AServer(
    agent=weather_agent,
    host="localhost",
    port=8000
)

# 启动服务器（使用任何ASGI服务器）
# 运行: uvicorn server:server

# client.py - 使用AG2创建A2A客户端
from ag2 import A2ARemoteAgent

# 连接到远程A2A智能体
remote_agent = A2ARemoteAgent(
    name="remote_weather",
    server_url="http://localhost:8000"
)

# 像使用本地智能体一样使用远程智能体
result = await remote_agent.generate_reply(
    messages=[{"role": "user", "content": "北京天气怎么样？"}]
)
print(result)

4.2 处理长时间运行的任务

A2A支持流式响应（Streaming）和异步任务，适合需要长时间处理的场景 ：

# 启用流式响应的智能体
async def handle_streaming_message(message, callback):
    """
    流式处理消息，逐步返回结果
    """
    # 第一步：发送开始处理状态
    await callback.send_status("processing", "开始分析数据...")
    
    # 模拟多步骤处理
    steps = ["获取数据", "分析模式", "生成报告", "完成"]
    for step in steps:
        await asyncio.sleep(2)  # 模拟耗时操作
        await callback.send_status("working", f"正在{step}...")
    
    # 最终结果
    await callback.send_result("分析完成，这是结果...")

4.3 安全性最佳实践

在生产环境中部署A2A智能体时，务必考虑以下安全措施 ：

# 带认证的Agent Card
agent_card = AgentCard(
    name="SecureAgent",
    # ... 其他配置
    authentication={
        "schemes": ["apiKey", "oauth2"],
        "credentials": {
            "apiKey": {
                "header": "X-API-Key",
                "location": "header"
            }
        }
    }
)

# 客户端携带API Key
headers = {
    "X-API-Key": "your-secure-api-key-here"
}
response = requests.get(agent_url, headers=headers)

4.4 常见问题与解决方案

问题	可能原因	解决方案
无法发现Agent Card	服务器未启动或路径错误	检查服务器是否运行，确认/.well-known/agent-card.json可访问
消息发送超时	网络问题或任务耗时过长	启用流式响应，或增加超时时间
跨框架通信失败	协议版本不兼容	确认双方使用相同版本的A2A协议
内存占用过高	未清理任务历史	配置LRU内存管理器，限制最大任务数

五、进阶实战：构建多智能体协作系统

案例3：旅行规划助手（多智能体协作）

场景：创建一个主控智能体，它需要协调多个专业智能体来完成完整的旅行规划任务：天气查询、景点推荐、酒店预订。

# travel_orchestrator.py
"""
旅行规划主控智能体
协调多个专业智能体完成任务
"""

import asyncio
from a2a_sdk import A2AClient, Message, TextPart
from a2a_sdk.discovery import DiscoveryService

class TravelOrchestrator:
    """
    旅行规划主控智能体
    负责发现专业智能体并协调任务
    """
    
    def __init__(self):
        self.discovery = DiscoveryService()
        self.agents = {}
        
    async def discover_specialized_agents(self):
        """
        发现所有可用的专业智能体
        """
        print("🔍 正在发现专业智能体...")
        
        # 假设这些是已注册的智能体地址
        agent_urls = {
            "weather": "http://localhost:8000",
            "attraction": "http://localhost:8001", 
            "hotel": "http://localhost:8002"
        }
        
        for agent_type, url in agent_urls.items():
            try:
                card = await self.discovery.discover_agent(url)
                if card:
                    self.agents[agent_type] = {
                        "url": url,
                        "card": card,
                        "client": A2AClient(base_url=url)
                    }
                    print(f"✅ 发现 {agent_type} 智能体: {card.get('name')}")
            except Exception as e:
                print(f"❌ 发现 {agent_type} 智能体失败: {e}")
    
    async def plan_trip(self, destination: str, days: int):
        """
        规划完整行程
        """
        print(f"\n🗺️ 开始规划 {destination} {days}天行程...\n")
        
        results = {}
        
        # 1. 查询天气
        if "weather" in self.agents:
            weather_msg = Message(
                role="user",
                parts=[TextPart(text=f"{destination}未来{days}天天气如何？")]
            )
            weather_resp = await self.agents["weather"]["client"].send_message(weather_msg)
            results["weather"] = self._extract_text(weather_resp)
            print(f"🌤️ 天气信息: {results['weather']}")
        
        # 2. 查询景点
        if "attraction" in self.agents:
            attraction_msg = Message(
                role="user",
                parts=[TextPart(text=f"{destination}必去的{3}个景点推荐")]
            )
            attraction_resp = await self.agents["attraction"]["client"].send_message(attraction_msg)
            results["attraction"] = self._extract_text(attraction_resp)
            print(f"🏞️ 景点推荐: {results['attraction']}")
        
        # 3. 查询酒店
        if "hotel" in self.agents:
            hotel_msg = Message(
                role="user",
                parts=[TextPart(text=f"{destination}性价比高的酒店推荐")]
            )
            hotel_resp = await self.agents["hotel"]["client"].send_message(hotel_msg)
            results["hotel"] = self._extract_text(hotel_resp)
            print(f"🏨 酒店推荐: {results['hotel']}")
        
        # 4. 生成综合报告
        print("\n" + "="*60)
        print("📋 行程规划总结")
        print("="*60)
        print(f"目的地: {destination}")
        print(f"行程天数: {days}天")
        print("\n".join([f"{k}: {v}" for k, v in results.items()]))
        
        return results
    
    def _extract_text(self, message):
        """从消息中提取文本内容"""
        for part in message.parts:
            if part.type == "text":
                return part.text
        return ""
    
    async def close(self):
        """关闭所有客户端连接"""
        for agent in self.agents.values():
            await agent["client"].close()

async def main():
    # 创建主控智能体
    orchestrator = TravelOrchestrator()
    
    # 发现专业智能体
    await orchestrator.discover_specialized_agents()
    
    # 规划行程
    await orchestrator.plan_trip("杭州", 3)
    
    # 关闭连接
    await orchestrator.close()

if __name__ == "__main__":
    asyncio.run(main())

六、热门文章要素优化

📌 关键词布局（SEO优化）

为了提升CSDN搜索排名，本文重点布局以下关键词：

• 核心关键词：A2A协议、Python智能体、AI智能体通信、多智能体协作
• 长尾关键词：Agent2Agent协议Python实现、智能体间通信标准、AI agent interoperability
• 品牌词：A2A Python SDK、Linux基金会A2A、Google A2A协议

📝 排版优化建议

1. 标题分级：使用H1主标题，H2/H3小节标题，符合CSDN文章结构
2. 代码块：所有代码包含语言标识和详细注释
3. 重点加粗：关键术语和技术名称加粗突出
4. 表格对比：用表格展示MCP vs A2A、核心组件等对比内容
5. 引用块：重要结论和注意事项使用引用格式

🎯 互动引导设计

在文章结尾加入以下互动元素：

1. 问题讨论：“你在开发AI智能体时遇到过哪些通信难题？欢迎在评论区分享！”
2. 资源分享：“需要本文完整代码的同学，私信回复’A2A源码’获取”
3. 下期预告：“下期将讲解《MCP协议实战：让AI智能体调用任何API》，敬请期待！”
4. 点赞收藏：“如果本文对你有帮助，请点赞👍收藏⭐支持一下！”

写在最后

A2A协议的诞生，标志着AI智能体从"单机游戏"走向"网络游戏"的时代已经来临 。作为Python开发者，掌握A2A协议将让你在构建复杂AI系统时游刃有余。

本文从零开始，带你了解了A2A的核心概念、环境配置、基础用法和进阶技巧，并提供了三个完整的实战案例。相信你已经可以开始用A2A构建自己的多智能体协作系统了。

记住核心要点：

• A2A让智能体学会"对话"，MCP让智能体拥有"手脚"
• Agent Card是智能体的"数字名片"
• 用Python SDK只需几行代码就能创建A2A智能体
• 多智能体协作可以像人类团队一样自然

如果你在实践过程中遇到任何问题，欢迎在评论区留言交流！

---

📚 参考资料：

1. Linux Foundation A2A Protocol Specification
2. A2A Python SDK Documentation
3. IBM A2A Tutorial Series
4. AG2 Framework A2A Integration Guide

🔗 相关资源：

• A2A GitHub仓库
• A2A官方文档
• Python Package Index - a2a-sdk

🏷️ 版权声明：本文为原创文章，转载请标明出处。