6. 与服务器交互¶

当 Helloworld A2A 服务器正在运行时，让我们向它发送一些请求。SDK 包含一个客户端（A2AClient），可以简化这些交互。

Helloworld 测试客户端¶

test_client.py 脚本演示了如何：

从服务器获取 Agent Card。
创建一个 A2AClient 实例。
发送非流式（message/send）和流式（message/stream）请求。

打开一个新的终端窗口，激活您的虚拟环境，并导航到 a2a-samples 目录。

激活虚拟环境（请确保在创建虚拟环境的同一目录下执行）：

Mac/LinuxWindows

source .venv/bin/activate

.venv\Scripts\activate

运行测试客户端：

# 从 a2a-samples 目录
python samples/python/agents/helloworld/test_client.py

理解客户端代码¶

让我们看看 test_client.py 的关键部分：

获取 Agent Card 并初始化客户端：

base_url = 'http://localhost:9999'

async with httpx.AsyncClient() as httpx_client:
    # Initialize A2ACardResolver
    resolver = A2ACardResolver(
        httpx_client=httpx_client,
        base_url=base_url,
        # agent_card_path uses default, extended_agent_card_path also uses default
    )

A2ACardResolver 类是一个便捷工具。它首先从服务器的 /.well-known/agent.json 端点（基于提供的 base URL）获取 AgentCard，然后用它初始化客户端。

发送非流式消息（send_message）：

client = A2AClient(
    httpx_client=httpx_client, agent_card=final_agent_card_to_use
)
logger.info('A2AClient initialized.')

send_message_payload: dict[str, Any] = {
    'message': {
        'role': 'user',
        'parts': [
            {'kind': 'text', 'text': 'how much is 10 USD in INR?'}
        ],
        'messageId': uuid4().hex,
    },
}
request = SendMessageRequest(
    id=str(uuid4()), params=MessageSendParams(**send_message_payload)
)

response = await client.send_message(request)
print(response.model_dump(mode='json', exclude_none=True))

send_message_payload 构建了 MessageSendParams 的数据。
这被包装在一个 SendMessageRequest 中。
它包含一个 message 对象，role 设置为 "user"，内容在 parts 中。
Helloworld agent 的 execute 方法会入队一条 "Hello World" 消息。DefaultRequestHandler 会检索并作为响应发送。
response 会是一个 SendMessageResponse 对象，其中包含 SendMessageSuccessResponse（agent 的 Message 作为结果）或 JSONRPCErrorResponse。

处理 Task ID（Helloworld 的说明性注释）：

Helloworld 客户端（test_client.py）不会直接尝试 get_task 或 cancel_task，因为简单的 Helloworld agent 的 execute 方法在通过 message/send 调用时，DefaultRequestHandler 会直接返回一个 Message 响应，而不是 Task 对象。更复杂的 agent（如 LangGraph 示例）会从 message/send 返回一个 Task 对象，其 id 可用于 get_task 或 cancel_task。
发送流式消息（send_message_streaming）：
```
streaming_request = SendStreamingMessageRequest(
    id=str(uuid4()), params=MessageSendParams(**send_message_payload)
)

stream_response = client.send_message_streaming(streaming_request)

async for chunk in stream_response:
    print(chunk.model_dump(mode='json', exclude_none=True))
```
- 此方法调用 agent 的 message/stream 端点。DefaultRequestHandler 会调用 HelloWorldAgentExecutor.execute 方法。
- execute 方法会入队一条 "Hello World" 消息，然后关闭事件队列。
- 客户端会收到这条消息作为一个 SendStreamingMessageResponse 事件，然后流结束。
- stream_response 是一个 AsyncGenerator。

预期输出¶

当您运行 test_client.py 时，您会看到如下 JSON 输出：

非流式响应（一条 "Hello World" 消息）。
流式响应（一条 "Hello World" 消息作为一个分片，之后流结束）。

id 字段在每次运行时都会不同。

// 非流式响应
{"jsonrpc":"2.0","id":"xxxxxxxx","result":{"type":"message","role":"agent","parts":[{"type":"text","text":"Hello World"}],"messageId":"yyyyyyyy"}}
// 流式响应（一个分片）
{"jsonrpc":"2.0","id":"zzzzzzzz","result":{"type":"message","role":"agent","parts":[{"type":"text","text":"Hello World"}],"messageId":"wwwwwwww","final":true}}

（实际的 xxxxxxxx、yyyyyyyy、zzzzzzzz、wwwwwwww 等 ID 会是不同的 UUID/请求 ID）

这表明您的服务器已能正确处理基于新版 SDK 结构的基本 A2A 交互！

现在，您可以在运行 __main__.py 的终端窗口中输入 Ctrl+C 来关闭服务器。