智能体通信协议

原文：Hello-Agents · 第十章智能体通信协议 MCP 等 · 在线阅读

一句话

智能体通信协议(MCP、A2A、ANP)解决了智能体与外部世界交互及多智能体协作的核心问题，提供了标准化的通信基础设施。

什么时候翻这页

需要让智能体访问外部工具和服务时
需要多个智能体相互协作完成任务时
需要构建大规模智能体网络时
想要构建自定义MCP服务器时

核心概念

智能体通信协议基础

MCP (Model Context Protocol)：由Anthropic团队提出，标准化智能体与外部工具/资源的通信方式，核心设计理念是"上下文共享"。

A2A (Agent-to-Agent Protocol)：由Google团队提出，实现智能体之间的点对点通信，核心设计理念是"对等通信"。

ANP (Agent Network Protocol)：构建大规模智能体网络的基础设施，核心设计理念是"去中心化服务发现"。

MCP协议架构

三层架构：

Host（宿主层）：负责接收用户提问并与模型交互
Client（客户端层）：负责与MCP Server建立连接，发送请求并接收响应
Server（服务器层）：执行具体的工具操作

MCP核心能力：

Tools：主动执行操作的工具
Resources：提供数据的被动资源
Prompts：提供指导性的提示模板

A2A协议特点

对等通信：每个智能体既是服务提供者，也是服务消费者。

消息类型：

task：任务分配
task_result：任务结果
negotiate：协商
vote：投票

ANP协议特点

核心概念：

Service：服务描述
Discovery：服务发现
Routing：智能路由
Load Balancing：负载均衡

网络拓扑：

星型拓扑
网状拓扑
分层拓扑

怎么做

使用MCP客户端

import asyncio
from hello_agents.protocols import MCPClient

async def connect_to_server():
    # 连接到MCP服务器
    client = MCPClient([
        "npx", "-y",
        "@modelcontextprotocol/server-filesystem",
        "."  # 指定根目录
    ])
    
    async with client:
        # 获取可用工具
        tools = await client.list_tools()
        print(f"可用工具: {[t['name'] for t in tools]}")
        
        # 调用工具
        result = await client.call_tool(
            "read_file",
            {"path": "README.md"}
        )
        print(f"文件内容: {result}")

asyncio.run(connect_to_server())

在智能体中使用MCP工具

from hello_agents import SimpleAgent, HelloAgentsLLM
from hello_agents.tools import MCPTool

agent = SimpleAgent(name="助手", llm=HelloAgentsLLM())

# 添加MCP工具
mcp_tool = MCPTool(
    name="filesystem",
    description="访问本地文件系统",
    server_command=["npx", "-y", "@modelcontextprotocol/server-filesystem", "."]
)
agent.add_tool(mcp_tool)

# 智能体可以直接使用展开后的工具
response = agent.run("读取README.md文件内容")
print(response)

使用A2A协议实现智能体协作

from hello_agents.protocols import A2AServer
import threading

# 创建技术专家Agent
tech_expert = A2AServer(
    name="技术专家",
    description="回答技术相关问题"
)

@tech_expert.skill("answer_question")
def answer_question(text: str) -> str:
    """回答技术问题"""
    # 处理技术问题的逻辑
    return f"技术回答：{text}"

# 启动服务
threading.Thread(target=lambda: tech_expert.run(port=6000), daemon=True).start()

# 创建接待员Agent
receptionist = SimpleAgent(
    name="接待员",
    llm=llm,
    system_prompt="你是客服接待员，负责分析问题并转给专家"
)

# 添加技术专家工具
tech_tool = A2ATool(
    agent_url="http://localhost:6000",
    name="tech_expert",
    description="技术专家"
)
receptionist.add_tool(tech_tool)

# 处理客户咨询
response = receptionist.run("你们的API如何调用？")
print(response)

使用ANP服务发现

from hello_agents.protocols import ANPDiscovery, register_service

# 创建服务发现中心
discovery = ANPDiscovery()

# 注册服务
register_service(
    discovery=discovery,
    service_id="nlp_agent_1",
    service_name="NLP处理专家",
    service_type="nlp",
    capabilities=["text_analysis", "sentiment_analysis"],
    endpoint="http://localhost:8001",
    metadata={"load": 0.3, "price": 0.01}
)

# 发现服务
nlp_services = discover_service(discovery, service_type="nlp")
print(f"找到 {len(nlp_services)} 个NLP服务")

构建自定义MCP服务器

#!/usr/bin/env python3
"""天气查询 MCP 服务器"""

import json
import requests
from hello_agents.protocols import MCPServer

# 创建 MCP 服务器
weather_server = MCPServer(name="weather-server", description="真实天气查询服务")

def get_weather(city: str) -> str:
    """获取指定城市的当前天气"""
    try:
        city_en = CITY_MAP.get(city, city)
        url = f"https://wttr.in/{city_en}?format=j1"
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        data = response.json()
        current = data["current_condition"][0]
        
        weather_data = {
            "city": city,
            "temperature": float(current["temp_C"]),
            "condition": current["weatherDesc"][0]["value"]
        }
        return json.dumps(weather_data, ensure_ascii=False)
    except Exception as e:
        return json.dumps({"error": str(e)}, ensure_ascii=False)

# 注册工具到服务器
weather_server.add_tool(get_weather)

if __name__ == "__main__":
    weather_server.run()

命令 / 代码速查

MCP工具使用

from hello_agents.tools import MCPTool

# 创建MCP工具
mcp_tool = MCPTool(
    name="calculator",
    server_command=["npx", "-y", "@modelcontextprotocol/server-github"]
)

# 在智能体中使用
agent.add_tool(mcp_tool)
response = agent.run("计算 25 乘以 16")

A2A工具使用

from hello_agents.protocols import A2AServer
from hello_agents.tools import A2ATool

# 创建A2A服务器
agent = A2AServer(name="专家", description="专业回答")

# 添加技能
@agent.skill("answer")
def answer(text: str) -> str:
    return f"回答: {text}"

# 启动服务
agent.run(port=5000)

# 创建客户端工具
a2a_tool = A2ATool(
    agent_url="http://localhost:5000",
    name="专家助手"
)

ANP工具使用

from hello_agents.protocols import ANPDiscovery
from hello_agents.tools import ANPTool

# 创建服务发现中心
discovery = ANPDiscovery()

# 注册服务
register_service(discovery, "service1", "服务1", "api", ["rest"], "http://localhost:8000")

# 创建ANP工具
anp_tool = ANPTool(discovery=discovery)
agent.add_tool(anp_tool)

与 Python / Claude Code 手册的联系

Python的异步编程(asyncio)用于MCP客户端的连接和工具调用
HTTP请求库(requests)用于MCP服务器的网络通信
JSON处理用于MCP协议的消息格式
多线程(threading)用于A2A服务器的并发处理
网络编程用于ANP协议的服务发现和路由

初学者易错点

MCP工具自动展开机制：添加MCPTool到智能体时，会自动将服务器提供的所有工具展开为独立工具，需要理解这一机制才能正确使用。
A2A服务启动顺序：在使用A2A协议时，必须先启动服务器Agent，再创建客户端工具，否则连接会失败。
ANP服务注册：使用ANP协议时，必须先注册服务，然后才能进行服务发现，顺序不能颠倒。
MCP传输方式选择：不同的传输方式适用于不同场景，错误的选择可能导致性能问题或连接失败。
自定义MCP服务器开发：开发自定义MCP服务器时，必须正确实现MCP协议规范，否则无法与客户端正常通信。

语义检索