6. 与服务器交互

Helloworld A2A 服务端运行起来之后，我们向它发送一些请求。

Helloworld 测试客户端

test_client.py 脚本演示了如何：

从服务端获取 Agent Card；
用 create_client 创建客户端；
发送非流式（SendMessage）与流式（SendStreamingMessage）请求。

打开一个新的终端窗口，激活虚拟环境（请在创建虚拟环境的同一目录下执行），并进入 a2a-samples 目录：

source .venv/bin/activate

.venv\Scripts\activate

运行测试客户端：

# from the a2a-samples directory
python samples/python/agents/helloworld/test_client.py

脚本的入口 main() 会开启一个交互会话，提示符为 user >：输入一句话就会向服务端发送一次非流式请求并打印响应，输入 exit 退出。文件中的 test_client_workflow 是一个 pytest 用例，会依次演示获取 Agent Card、非流式调用、流式调用和获取扩展 Agent Card 四个环节，并通过 fixture 自动启动服务端。

理解客户端代码

来看 test_client.py 的几个关键部分。

1. 获取 Agent Card

import httpx

from a2a.client import A2ACardResolver

# Initializes the A2ACardResolver instance with an HTTP client, base URL,
# and uses the default path for the agent card.
async with httpx.AsyncClient() as httpx_client:
    resolver = A2ACardResolver(
        httpx_client=httpx_client,
        base_url='http://127.0.0.1:9999',
        # Provide agent_card_path, if your agent uses a different path
        # agent_card_path=''
    )
    public_agent_card = await resolver.get_agent_card()

A2ACardResolver 是一个便利类。调用 get_agent_card() 时，它会根据提供的基础 URL，从服务端的 /.well-known/agent-card.json 端点获取 AgentCard，随后用它来初始化客户端。

2. 初始化客户端并发送非流式消息

from a2a.client import ClientConfig, create_client
from a2a.helpers import new_text_message
from a2a.types import Role, SendMessageRequest

print('\nInitializing a non-streaming client.')
config = ClientConfig(streaming=False)
client = await create_client(agent=public_agent_card, client_config=config)

# Creates a new text message to be sent to the A2A Server.
# Ex: text_query = 'Why is the sky blue?'
message = new_text_message(text_query, role=Role.ROLE_USER)
request = SendMessageRequest(message=message)

print('Response:')
async for chunk in client.send_message(request):
    print(chunk)

create_client 函数根据 AgentCard 提供的信息与 ClientConfig 创建一个 Client；
用 new_text_message 辅助函数构造 Message（传入 role=Role.ROLE_USER），再包装进 SendMessageRequest；
客户端的 send_message 方法返回一个异步迭代器，它会产出智能体的最终 Task 或 Message 响应。在本例中是一个 Task。

3. 初始化客户端并发送流式消息

print('\nInitializing a streaming client.')
client_config = ClientConfig(streaming=True)  # Streaming
client = await create_client(agent=public_agent_card, client_config=client_config)

print('Response:')
async for chunk in client.send_message(request):
    print(chunk)

通过在 ClientConfig 中设置 streaming=True，用 create_client 创建一个独立的流式客户端；
再次调用 send_message，此时它会以流的方式产出事件：循环的每一次迭代都打印一个随网络到达的数据块；
循环结束后调用 await client.close() 释放底层的 HTTP 连接。

预期的输出

结合 Helloworld 执行器的逻辑，你会看到：

公开 Agent Card：以格式化摘要展示 Hello World Agent 的基本信息、接口、能力、输入输出模式和技能；
非流式响应：单个以 protobuf 文本格式打印的 Task，其状态为 TASK_STATE_COMPLETED，包含一个带有问候文本的产出物，历史中有用户消息和智能体的中间状态消息 “Processing request…”；
流式响应：依次到达的数据块——初始的 task（状态为 TASK_STATE_SUBMITTED）、状态为 TASK_STATE_WORKING 的 status_update、带有结果的 artifact_update，以及状态为 TASK_STATE_COMPLETED 的最终 status_update；
扩展 Agent Card：在公开卡片的基础上多出 echo_bot_super_mode 技能（通过 GetExtendedAgentCard 获取）。

输出中的 id 字段每次运行都会不同。看到上述结果，说明服务端已经能正确处理基础的 A2A 交互。完成后，在运行 __main__.py 的终端中按 Ctrl+C 即可关闭服务端。

下一步

Helloworld 演示了 A2A 的基本机制。接下来通过 LangGraph 示例了解流式传输与多轮交互。

来源：A2A 官方文档 · Interacting with the Server

5. 启动服务器 7. 流式传输与多轮交互（LangGraph 示例）