目录

• • 介绍
• • 设置环境
• • 创建项目
• • 代理技能
• • 代理卡片
• • A2A服务器
• • 与A2A服务器交互
• • 添加代理功能
• • 使用本地Ollama模型
• • 后续步骤

介绍

在本教程中，您将使用Python构建一个简单的echo A2A服务器。这个基础实现将向您展示A2A提供的所有功能。完成本教程后，您将能够使用Ollama或Google的Agent Development Kit添加代理功能。

您将学习：

• • A2A背后的基本概念
• • 如何用Python创建A2A服务器
• • 与A2A服务器交互
• • 添加训练模型作为代理

设置环境

您需要的工具

• • 代码编辑器，如Visual Studio Code (VS Code)
• • 命令提示符，如Terminal (Linux)、iTerm (Mac) 或VS Code中的Terminal

Python环境

我们将使用uv作为包管理器并设置项目。

我们将使用的A2A库需要python >= 3.12，如果您还没有匹配的版本，uv可以安装。我们将使用python 3.12。

检查

运行以下命令，确保您已准备好进入下一步：

echo 'import sys; print(sys.version)' | uv run -

如果您看到类似以下内容，说明您已准备就绪！

3.12.3 (main, Feb 4 2025, 14:48:35) [GCC 13.3.0]

创建项目

首先使用uv创建一个项目。我们将添加--package标志，以便您以后可以添加测试或发布项目：

uv init --package my-project
cd my-project

使用虚拟环境

我们为这个项目创建一个虚拟环境。这只需要做一次：

uv venv .venv

对于这个和将来打开的任何终端窗口，您需要激活这个虚拟环境：

source .venv/bin/activate

如果您使用的是VS Code等代码编辑器，您需要设置Python解释器以便代码补全。在VS Code中，按下Ctrl-Shift-P并选择Python: Select Interpreter。然后选择您的项目my-project，接着选择正确的Python解释器Python 3.12.3 ('.venv':venv) ./.venv/bin/python

现在源代码结构应该类似于：

tree .
.
├── pyproject.toml
├── README.md
├── src
│   └── my-project
│       ├── __init__.py

添加Google-A2A Python库

接下来我们将添加来自Google的A2A Python示例库：

uv add git+https://github.com/google/A2A#subdirectory=samples/python

设置项目结构

现在创建一些我们稍后将使用的文件：

touch src/my_project/agent.py
touch src/my_project/task_manager.py

测试运行

如果一切设置正确，您现在应该能够运行您的应用程序：

uv run my-project

输出应该类似于：

Hello from my-project!

代理技能

代理技能是代理可以执行的一组功能。下面是我们echo代理的技能示例：

{id: "my-project-echo-skill"name: "Echo Tool",description: "Echos the input given",tags: ["echo", "repeater"],examples: ["I will see this echoed back to me"],inputModes: ["text"],outputModes: ["text"]
}

这符合代理卡片的技能部分：

{id: string; // 代理技能的唯一标识符name: string; // 技能的人类可读名称// 技能描述 - 将被客户端或人类用作提示，以理解这个技能的作用description: string;// 描述这个特定技能功能类别的标签词集合// (例如"cooking"、"customer support"、"billing")tags: string[];// 该技能可以执行的示例场景集合// 将被客户端用作提示，以了解如何使用该技能// (例如"I need a recipe for bread")examples?: string[]; // 任务提示示例// 该技能支持的交互模式集合// (如果与默认值不同)inputModes?: string[]; // 支持的输入MIME类型outputModes?: string[]; // 支持的输出MIME类型
}

实现

让我们用代码创建这个代理技能。打开src/my-project/__init__.py并用以下代码替换内容：

import google_a2a
from google_a2a.common.types import AgentSkilldef main():skill = AgentSkill(id="my-project-echo-skill",name="Echo Tool",description="Echos the input given",tags=["echo", "repeater"],examples=["I will see this echoed back to me"],inputModes=["text"],outputModes=["text"],)print(skill)if __name__ == "__main__":main()

测试运行

让我们运行一下：

uv run my-project

输出应该类似于：

id='my-project-echo-skill' name='Echo Tool' description='Echos the input given' tags=['echo', 'repeater'] examples=['I will see this echoed back to me'] inputModes=['text'] outputModes=['text']

代理卡片

现在我们已经定义了技能，可以创建代理卡片了。

远程代理需要以JSON格式发布代理卡片，描述代理的能力和技能，以及认证机制。换句话说，这让世界了解您的代理及如何与之交互。

实现

首先添加一些解析命令行参数的辅助工具。这对稍后启动服务器很有帮助：

uv add click

然后更新我们的代码：

import loggingimport click
from dotenv import load_dotenv
import google_a2a
from google_a2a.common.types import AgentSkill, AgentCapabilities, AgentCardlogging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)@click.command()
@click.option("--host", default="localhost")
@click.option("--port", default=10002)
def main(host, port):skill = AgentSkill(id="my-project-echo-skill",name="Echo Tool",description="Echos the input given",tags=["echo", "repeater"],examples=["I will see this echoed back to me"],inputModes=["text"],outputModes=["text"],)logging.info(skill)if __name__ == "__main__":main()

接下来添加我们的代理卡片：

# ...
def main(host, port):# ...capabilities = AgentCapabilities()agent_card = AgentCard(name="Echo Agent",description="This agent echos the input given",url=f"http://{host}:{port}/",version="0.1.0",defaultInputModes=["text"],defaultOutputModes=["text"],capabilities=capabilities,skills=[skill])logging.info(agent_card)if __name__ == "__main__":main()

测试运行

让我们运行一下：

uv run my-project

输出应该类似于：

INFO:root:name='Echo Agent' description='This agent echos the input given' url='http://localhost:10002/' provider=None version='0.1.0' documentationUrl=None capabilities=AgentCapabilities(streaming=False, pushNotifications=False, stateTransitionHistory=False) authentication=None defaultInputModes=['text'] defaultOutputModes=['text'] skills=[AgentSkill(id='my-project-echo-skill', name='Echo Tool', description='Echos the input given', tags=['echo', 'repeater'], examples=['I will see this echoed back to me'], inputModes=['text'], outputModes=['text'])]

A2A服务器

我们几乎准备好启动服务器了！我们将使用Google-A2A中的A2AServer类，它在底层启动一个uvicorn服务器。

任务管理器

在创建服务器之前，我们需要一个任务管理器来处理传入的请求。

我们将实现InMemoryTaskManager接口，需要实现两个方法：

async def on_send_task(self,request: SendTaskRequest
) -> SendTaskResponse:"""该方法查询或创建代理的任务。调用者将收到恰好一个响应。"""passasync def on_send_task_subscribe(self,request: SendTaskStreamingRequest
) -> AsyncIterable[SendTaskStreamingResponse] | JSONRPCResponse:"""该方法使调用者订阅有关任务的未来更新。调用者将收到一个响应，并通过客户端和服务器之间建立的会话接收订阅更新"""pass

打开src/my_project/task_manager.py并添加以下代码。我们将简单地返回直接回显响应，并立即将任务标记为完成，不需要任何会话或订阅：

from typing import AsyncIterableimport google_a2a
from google_a2a.common.server.task_manager import InMemoryTaskManager
from google_a2a.common.types import (Artifact,JSONRPCResponse,Message,SendTaskRequest,SendTaskResponse,SendTaskStreamingRequest,SendTaskStreamingResponse,Task,TaskState,TaskStatus,TaskStatusUpdateEvent,
)class MyAgentTaskManager(InMemoryTaskManager):def __init__(self):super().__init__()async def on_send_task(self, request: SendTaskRequest) -> SendTaskResponse:# 更新由InMemoryTaskManager存储的任务await self.upsert_task(request.params)task_id = request.params.id# 我们的自定义逻辑，简单地将任务标记为完成# 并返回echo文本received_text = request.params.message.parts[0].texttask = await self._update_task(task_id=task_id,task_state=TaskState.COMPLETED,response_text=f"on_send_task received: {received_text}")# 发送响应return SendTaskResponse(id=request.id, result=task)async def on_send_task_subscribe(self,request: SendTaskStreamingRequest) -> AsyncIterable[SendTaskStreamingResponse] | JSONRPCResponse:passasync def _update_task(self,task_id: str,task_state: TaskState,response_text: str,) -> Task:task = self.tasks[task_id]agent_response_parts = [{"type": "text","text": response_text,}]task.status = TaskStatus(state=task_state,message=Message(role="agent",parts=agent_response_parts,))task.artifacts = [Artifact(parts=agent_response_parts,)]return task

A2A服务器

有了任务管理器，我们现在可以创建服务器了。

打开src/my_project/__init__.py并添加以下代码：

# ...
from google_a2a.common.server import A2AServer
from my_project.task_manager import MyAgentTaskManager
# ...
def main(host, port):# ...task_manager = MyAgentTaskManager()server = A2AServer(agent_card=agent_card,task_manager=task_manager,host=host,port=port,)server.start()

测试运行

让我们运行一下：

uv run my-project

输出应该类似于：

INFO:     Started server process [20506]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:10002 (Press CTRL+C to quit)

恭喜！您的A2A服务器现在正在运行！

与A2A服务器交互

首先我们将使用Google-A2A的命令行工具向我们的A2A服务器发送请求。尝试之后，我们将编写自己的基本客户端，了解底层工作原理。

使用Google-A2A的命令行工具

在上一步中，您的A2A服务器已经在运行：

# 这应该已经在您的终端中运行
$ uv run my-project
INFO:     Started server process [20538]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:10002 (Press CTRL+C to quit)

在同一目录中打开新终端：

source .venv/bin/activate
uv run google-a2a-cli --agent http://localhost:10002

注意：这只有在您安装了来自此PR的google-a2a时才有效，因为之前CLI并未公开。

否则，您必须直接检出Google/A2A仓库，导航到samples/python目录并直接运行CLI。

然后通过输入并按Enter发送消息到服务器：

=========  starting a new task ========What do you want to send to the agent? (:q or quit to exit): Hello!

如果一切正常，您将在响应中看到：

"message":{"role":"agent","parts":[{"type":"text","text":"on_send_task received: Hello!"}]}

要退出，输入:q并按Enter。

添加代理功能

现在我们有了一个基本的A2A服务器，让我们添加更多功能。我们将探索A2A如何异步工作和流式响应。

流式传输

这允许客户端订阅服务器并接收多个更新，而不是单个响应。这对于长时间运行的代理任务或需要向客户端流式传输多个Artifacts的情况很有用。

首先声明我们的代理已准备好流式传输。打开src/my_project/__init__.py并更新AgentCapabilities：

# ...
def main(host, port):# ...capabilities = AgentCapabilities(streaming=True)# ...

现在在src/my_project/task_manager.py中，我们需要实现on_send_task_subscribe：

import asyncio
# ...
class MyAgentTaskManager(InMemoryTaskManager):# ...async def _stream_3_messages(self, request: SendTaskStreamingRequest):task_id = request.params.idreceived_text = request.params.message.parts[0].texttext_messages = ["one", "two", "three"]for text in text_messages:parts = [{"type": "text","text": f"{received_text}: {text}",}]message = Message(role="agent", parts=parts)is_last = text == text_messages[-1]task_state = TaskState.COMPLETED if is_last else TaskState.WORKINGtask_status = TaskStatus(state=task_state,message=message)task_update_event = TaskStatusUpdateEvent(id=request.params.id,status=task_status,final=is_last,)await self.enqueue_events_for_sse(request.params.id,task_update_event)async def on_send_task_subscribe(self,request: SendTaskStreamingRequest) -> AsyncIterable[SendTaskStreamingResponse] | JSONRPCResponse:# 更新由InMemoryTaskManager存储的任务await self.upsert_task(request.params)task_id = request.params.id# 为此任务创建工作队列sse_event_queue = await self.setup_sse_consumer(task_id=task_id)# 开始为此任务异步工作asyncio.create_task(self._stream_3_messages(request))# 告诉客户端期待未来的流式响应return self.dequeue_events_for_sse(request_id=request.id,task_id=task_id,sse_event_queue=sse_event_queue,)

重启A2A服务器以应用新更改，然后重新运行CLI：

$ uv run google-a2a-cli --agent http://localhost:10002
=========  starting a new task ========What do you want to send to the agent? (:q or quit to exit): Streaming?"status":{"state":"working","message":{"role":"agent","parts":[{"type":"text","text":"Streaming?: one"}]}
"status":{"state":"working","message":{"role":"agent","parts":[{"type":"text","text":"Streaming?: two"}]}
"status":{"state":"completed","message":{"role":"agent","parts":[{"type":"text","text":"Streaming?: three"}]}

有时代理可能需要额外输入。例如，代理可能会询问客户是否希望继续重复3条消息。在这种情况下，代理将以TaskState.INPUT_REQUIRED响应，客户端然后会用相同的task_id和session_id但更新的消息重新发送send_task_streaming，提供代理所需的输入。在服务器端，我们将更新on_send_task_subscribe以处理这种情况：

# ...class MyAgentTaskManager(InMemoryTaskManager):# ...async def _stream_3_messages(self, request: SendTaskStreamingRequest):# ...async for message in messages:# ...# is_last = message == messages[-1] # 删除此行task_state = TaskState.WORKING# ...task_update_event = TaskStatusUpdateEvent(id=request.params.id,status=task_status,final=False,)# ...ask_message = Message(role="agent",parts=[{"type": "text","text": "Would you like more messages? (Y/N)"}])task_update_event = TaskStatusUpdateEvent(id=request.params.id,status=TaskStatus(state=TaskState.INPUT_REQUIRED,message=ask_message),final=True,)await self.enqueue_events_for_sse(request.params.id,task_update_event)# ...async def on_send_task_subscribe(self,request: SendTaskStreamingRequest) -> AsyncIterable[SendTaskStreamingResponse] | JSONRPCResponse:task_id = request.params.idis_new_task = task_id in self.tasks# 更新由InMemoryTaskManager存储的任务await self.upsert_task(request.params)received_text = request.params.message.parts[0].textsse_event_queue = await self.setup_sse_consumer(task_id=task_id)if not is_new_task and received_text == "N":task_update_event = TaskStatusUpdateEvent(id=request.params.id,status=TaskStatus(state=TaskState.COMPLETED,message=Message(role="agent",parts=[{"type": "text","text": "All done!"}])),final=True,)await self.enqueue_events_for_sse(request.params.id,task_update_event,)else:asyncio.create_task(self._stream_3_messages(request))return self.dequeue_events_for_sse(request_id=request.id,task_id=task_id,sse_event_queue=sse_event_queue,)

重启服务器并运行CLI后，我们可以看到任务将继续运行，直到我们告诉代理N：

$ uv run google-a2a-cli --agent http://localhost:10002
=========  starting a new task ========What do you want to send to the agent? (:q or quit to exit): Streaming?"status":{"state":"working","message":{"role":"agent","parts":[{"type":"text","text":"Streaming?: one"}]}
"status":{"state":"working","message":{"role":"agent","parts":[{"type":"text","text":"Streaming?: two"}]}
"status":{"state":"working","message":{"role":"agent","parts":[{"type":"text","text":"Streaming?: three"}]}
"status":{"state":"input-required","message":{"role":"agent","parts":[{"type":"text","text":"Would you like more messages? (Y/N)"}]}What do you want to send to the agent? (:q or quit to exit): N"status":{"state":"completed","message":{"role":"agent","parts":[{"type":"text","text":"All done!"}]}

恭喜！您现在有了一个能够异步执行工作并在需要时向用户请求输入的代理。

使用本地Ollama模型

现在到了激动人心的部分。我们将为A2A服务器添加AI功能。

在本教程中，我们将设置本地Ollama模型并将其与A2A服务器集成。

要求

我们将安装ollama、langchain，并下载支持MCP工具的ollama模型(用于未来教程)。

1. 1. 下载ollama
2. 2. 运行ollama服务器：

# 注意：如果ollama已在运行，您可能会收到错误，如
# Error: listen tcp 127.0.0.1:11434: bind: address already in use
# 在Linux上可以运行systemctl stop ollama停止ollama
ollama serve

1. 3. 从此列表下载模型。我们将使用qwq，因为它支持tools(如其标签所示)并在24GB显卡上运行：

ollama pull qwq

1. 4. 安装langchain：

uv add langchain langchain-ollama langgraph

现在ollama设置好了，我们可以开始将其集成到A2A服务器中。

将Ollama集成到A2A服务器

首先打开src/my_project/__init__.py：

# ...@click.command()
@click.option("--host", default="localhost")
@click.option("--port", default=10002)
@click.option("--ollama-host", default="http://127.0.0.1:11434")
@click.option("--ollama-model", default=None)
def main(host, port, ollama_host, ollama_model):# ...capabilities = AgentCapabilities(streaming=False # 我们将流式功能作为读者的练习)# ...task_manager = MyAgentTaskManager(ollama_host=ollama_host,ollama_model=ollama_model,)# ..

现在在src/my_project/agent.py中添加AI功能：

from langchain_ollama import ChatOllama
from langgraph.prebuilt import create_react_agent
from langgraph.graph.graph import CompiledGraphdef create_ollama_agent(ollama_base_url: str, ollama_model: str):ollama_chat_llm = ChatOllama(base_url=ollama_base_url,model=ollama_model,temperature=0.2)agent = create_react_agent(ollama_chat_llm, tools=[])return agentasync def run_ollama(ollama_agent: CompiledGraph, prompt: str):agent_response = await ollama_agent.ainvoke({"messages": prompt })message = agent_response["messages"][-1].contentreturn str(message)

最后从src/my_project/task_manager.py调用我们的ollama代理：

# ...
from my_project.agent import create_ollama_agent, run_ollamaclass MyAgentTaskManager(InMemoryTaskManager):def __init__(self,ollama_host: str,ollama_model: typing.Union[None, str]):super().__init__()if ollama_model is not None:self.ollama_agent = create_ollama_agent(ollama_base_url=ollama_host,ollama_model=ollama_model)else:self.ollama_agent = Noneasync def on_send_task(self, request: SendTaskRequest) -> SendTaskResponse:# ...received_text = request.params.message.parts[0].textresponse_text = f"on_send_task received: {received_text}"if self.ollama_agent is not None:response_text = await run_ollama(ollama_agent=self.ollama_agent, prompt=received_text)task = await self._update_task(task_id=task_id,task_state=TaskState.COMPLETED,response_text=response_text)# 发送响应return SendTaskResponse(id=request.id, result=task)# ...

让我们测试一下！

首先重新运行A2A服务器，将qwq替换为您下载的ollama模型：

uv run my-project --ollama-host http://127.0.0.1:11434 --ollama-model qwq

然后重新运行CLI：

uv run google-a2a-cli --agent http://localhost:10002

注意，如果您使用大模型，加载可能需要一段时间。CLI可能会超时。在这种情况下，一旦ollama服务器完成模型加载，请重新运行CLI。

您应该看到类似于以下内容：

=========  starting a new task ========What do you want to send to the agent? (:q or quit to exit): hey"message":{"role":"agent","parts":[{"type":"text","text":"<think>\nOkay, the user said \"hey\". That's pretty casual. I should respond in a friendly way. Maybe ask how I can help them today. Keep it open-ended so they feel comfortable sharing what they need. Let me make sure my tone is positive and approachable. Alright, something like, \"Hey there! How can I assist you today?\" Yeah, that sounds good.\n</think>\n\nHey there! How can I assist you today? 😊"}]}

恭喜！您现在有了一个使用AI模型生成响应的A2A服务器！

了解更多：https://a2aprotocol.ai/blog/python-a2a-tutorial