Agno：A2A/AG-UI/MCP

AI编程/Vibe Coding 遇到问题需要帮助的，联系微信 ezpoda，免费咨询。

MCP 将智能体连接到工具和数据，A2A 让智能体之间互相交流，AG-UI 将丰富的事件流式传输到你的前端，Agno 将这三者干净地连接起来，使你能够构建可互操作、实时的智能体应用。

智能体系统终于开始收敛到开放协议，而不是定制的胶水代码。

实际上，我们需要三个部分：

• MCP（模型上下文协议） 将智能体连接到工具、数据和上下文。
• A2A（智能体到智能体） 让智能体能够在不同框架之间发现、消息传递和协作。
• AG-UI（智能体-用户交互） 在智能体和前端之间流式传输结构化的、实时的事件。

Agno 的 AgentOS 并排暴露了 A2A 和 AG-UI，赋予我们与其他智能体的互操作性和与 UI 的干净流式接口，同时我们继续使用 MCP 进行工具/数据访问。

为什么这很重要：

• 互操作性： 使用 A2A 将工作委托给用其他技术栈（LangGraph、CrewAI、ADK、自定义）构建的智能体，而无需一次性 API。规范定义的智能体卡片和流式传输使其变得可预测。
• 可靠的前端： AG-UI 标准化事件名称和流程（文本增量、工具调用、状态增量），因此 UI 可以简单且可调试，而不是临时的 WebSockets。
• 更简单的心理模型： MCP 用于工具/数据，A2A 用于智能体↔智能体，AG-UI 用于智能体↔UI。不会混淆哪个部分做任何事情。

1、协议栈概览

• MCP（模型上下文协议）

开放标准，用于将 AI 应用/智能体与工具、数据源和上下文连接（可以想象成 AI 应用的 USB-C）。 使用它来挂载 GitHub、数据库、文档、计算器等。

• A2A（智能体到智能体）

智能体之间用于发现、能力描述、任务/流式传输和安全性的开放标准。 该规范正式定义了智能体卡片、JSON-RPC/gRPC/HTTP+JSON 传输和 SSE 流式传输模式。

• AG-UI（智能体-用户交互）

用于前端↔智能体通信的开放的、基于事件的协议。 它定义了规范的事件名称（例如，RunStarted、TextMessageContent、ToolCallStart、StateDelta）和你的 UI 可以通过 HTTP（SSE/NDJSON）使用的事件流。

2、使用 Agno 的 A2A

A2A 规范建议在众所周知的路径托管公共智能体卡片： https://{domain}/.well-known/agent-card.json

该卡片描述身份、能力、技能、端点/传输和安全方案。也可以使用经过身份验证的"扩展"卡片。

在 AgentOS 上启用 A2A，你会获得 JSON-RPC 流加流式传输：

• POST /a2a/message/send：发送一条消息（JSON-RPC 响应包含一个任务）
• POST /a2a/message/stream：带有实时更新（状态、消息、工件）的NDJSON流

Agno 的参考文档记录了这两个端点，并确认了 NDJSON 流式传输。

最小服务器：

from agno.agent import Agent
from agno.os import AgentOS
from agno.os.interfaces.a2a import A2A

agent = Agent(name="Research Agent")

agent_os = AgentOS(agents=[agent], interfaces=[A2A(agents=[agent])])

app = agent_os.get_app()

if __name__ == "__main__":
    agent_os.serve(app="a2a_example:app", host="0.0.0.0", port=8000, reload=True)

符合规范的客户端（发送 + 流式传输）：

import asyncio, json, httpx

BASE = "http://localhost:8000"

async def a2a_send_and_stream(agent_id: str, text: str):
    async with httpx.AsyncClient(timeout=None) as client:
        # 1) 发送
        send_resp = await client.post(
            f"{BASE}/a2a/message/send",
            json={"message": {"agentId": agent_id, "parts": [{"kind": "text", "text": text}]}}
        )
        send_resp.raise_for_status()
        task_id = send_resp.json()["result"]["task"]["id"]

        # 2) 流式传输 NDJSON 更新
        async with client.stream("POST", f"{BASE}/a2a/message/stream", json={"taskId": task_id}) as stream:
            async for line in stream.aiter_lines():
                if line:
                    print("event:", json.loads(line))

asyncio.run(a2a_send_and_stream("Research Agent", "Summarise RBI news"))

3、使用 Agno 的 AG-UI

AG-UI 定义了 PascalCase 中的统一事件分类，包括：

• 生命周期： RunStarted、RunFinished、RunError
• 文本： TextMessageStart、TextMessageContent、TextMessageEnd
• 工具调用： ToolCallStart、ToolCallArgs、ToolCallEnd、ToolCallResult
• 状态： StateSnapshot、StateDelta

这些事件按顺序流式传输，并且很容易在 UI 中逐步渲染。

通过以下方式暴露你的智能体：

• POST /agui — 主入口点；接受标准 AG-UI 负载并流式传输事件
• GET /status — 健康/能力

Agno 的文档还展示了如何运行 Dojo，即即用型 AG-UI 前端，用于快速测试。

最小服务器：

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.os import AgentOS
from agno.os.interfaces.agui import AGUI

chat_agent = Agent(
    name="Support Agent",
    model=OpenAIChat(id="gpt-4o"),
    instructions=["Be concise and clarify when needed."],
    markdown=True,
)

agent_os = AgentOS(agents=[chat_agent], interfaces=[AGUI(agent=chat_agent)])

app = agent_os.get_app()

if __name__ == "__main__":
    agent_os.serve(app="agui_example:app", host="0.0.0.0", port=7777, reload=True)

带有规范事件名称的微型 React 挂钩（SSE）：

export async function streamAgui(agentUrl: string, payload: unknown, onEvent: (e:any)=>void) {
  const resp = await fetch(`${agentUrl}/agui`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', 'Accept': 'text/event-stream' },
    body: JSON.stringify(payload),
  });
  const reader = resp.body!.getReader();
  const dec = new TextDecoder();
  let buf = '';
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buf += dec.decode(value, { stream: true });
    for (const line of buf.split('\n')) {
      if (line.startsWith('data: ')) onEvent(JSON.parse(line.slice(6)));
    }
    buf = '';
  }
}

3、同时运行两个协议

Agno 让你可以在同一个进程中挂载A2A 和 AG-UI，因此智能体可以与对等节点协作并同时流式传输到你的 UI。

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.os import AgentOS
from agno.os.interfaces.a2a import A2A
from agno.os.interfaces.agui import AGUI


research_agent = Agent(
    name="Multi-Protocol Research Agent",
    model=OpenAIChat(id="gpt-4o"),
    instructions=["Do thorough research and stream progress updates."],
    markdown=True,
)

agent_os = AgentOS(
    agents=[research_agent],
    interfaces=[A2A(agents=[research_agent]), AGUI(agent=research_agent)]
)

app = agent_os.get_app()

if __name__ == "__main__":
    agent_os.serve(app="multi_proto:app", host="0.0.0.0", port=8000, reload=True)

实用模式：

• 中心辐射模型： 通过 AG-UI 暴露编排器；让它通过 A2A 委托给专家（搜索、金融、RAG）。专家可以是 Agno 或其他框架，只要它们符合 A2A 标准即可。
• 对等网状模型： 多个团队使用 A2A 通信，你的 React 应用通过 AG-UI 连接到任何一个。
• 内部 MCP： 智能体使用 MCP 服务器获取数据/工具，而 A2A 仍然是网络上的智能体↔智能体协议。

4、结束语

我们不需要专有的连接来构建严肃的智能体系统。使用 MCP 进行工具和上下文，使用 A2A 进行智能体到智能体协作，使用 AG-UI 交付稳定的、流式传输的 UI 合约。

通过 Agno 的 AgentOS，我们可以一起运行 A2A 和 AG-UI，并发布可互操作、实时和可用于生产的东西。

原文链接: Building the Future of Agent Communication: Agno's A2A & AG-UI Integration (with MCP in the mix)

汇智网翻译整理，转载请标明出处