6. 与服务器交互¶

在 Helloworld A2A 服务器运行后，我们向它发送一些请求。SDK 包含一个客户端 (A2AClient)，可以简化这些交互。

Helloworld 测试客户端¶

test_client.py 脚本演示了如何

从服务器获取代理卡。
创建 A2AClient 实例。
发送非流式 (message/send) 和流式 (message/stream) 请求。

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

激活虚拟环境（请务必在创建虚拟环境的同一目录中执行此操作）

Mac/LinuxWindows

source .venv/bin/activate

.venv\Scripts\activate

运行测试客户端

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

理解客户端代码¶

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

获取代理卡和初始化客户端:

base_url = 'https://: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-card.json 端点（基于提供的基本 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 代理的 execute 方法将排队一个“Hello World”消息。DefaultRequestHandler 将检索此消息并将其作为响应发送。
response 将是一个 SendMessageResponse 对象，其中包含一个 SendMessageSuccessResponse（以代理的 Message 作为结果）或一个 JSONRPCErrorResponse。

处理任务 ID（Helloworld 的说明性注释）:

Helloworld 客户端 (test_client.py) 不会直接尝试 get_task 或 cancel_task，因为当通过 message/send 调用时，简单的 Helloworld 代理的 execute 方法会导致 DefaultRequestHandler 返回直接的 Message 响应，而不是 Task 对象。更复杂的代理（如 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))
```
- 此方法调用代理的 message/stream 端点。DefaultRequestHandler 将调用 HelloWorldAgentExecutor.execute 方法。
- execute 方法排队一个“Hello World”消息，然后事件队列关闭。
- 客户端将接收此单个消息作为一次 SendStreamingMessageResponse 事件，然后流将终止。
- stream_response 是一个 AsyncGenerator。

预期输出¶

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

非流式响应（一条“Hello World”消息）。
流式响应（一条“Hello World”消息作为一段，之后流结束）。

输出中的 id 字段每次运行都会有所不同。

// Non-streaming response
{"jsonrpc":"2.0","id":"xxxxxxxx","result":{"type":"message","role":"agent","parts":[{"type":"text","text":"Hello World"}],"messageId":"yyyyyyyy"}}
// Streaming response (one chunk)
{"jsonrpc":"2.0","id":"zzzzzzzz","result":{"type":"message","role":"agent","parts":[{"type":"text","text":"Hello World"}],"messageId":"wwwwwwww","final":true}}

（实际 ID，如 xxxxxxxx、yyyyyyyy、zzzzzzzz、wwwwwwww 将是不同的 UUID/请求 ID）

这证实您的服务器正在使用更新后的 SDK 结构正确处理基本的 A2A 交互！

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