Skip to content

6. 与服务器交互

随着 Helloworld A2A 服务器的运行,让我们向它发送一些请求。SDK 包含一个客户端(A2AClient),简化了这些交互。

Helloworld 测试客户端

test_client.py 脚本演示了如何:

  1. 从服务器获取 Agent Card。
  2. 创建 A2AClient 实例。
  3. 发送非流式传输(message/send)和流式传输(message/stream)请求。

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

激活虚拟环境(请确保在创建虚拟环境的同一目录中执行此操作):

source .venv/bin/activate
.venv\Scripts\activate

运行测试客户端:

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

理解客户端代码

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

  1. 获取 Agent Card 和初始化客户端

    
    

    A2ACardResolver 类是一个便利工具。它首先从服务器的 /.well-known/agent-card.json 端点(基于提供的基础 URL)获取 AgentCard,然后用它初始化客户端。

  2. 发送非流式消息(send_message

    
    
    • send_message_payload 构造 MessageSendParams 的数据。
    • 这被包装在 SendMessageRequest 中。
    • 它包含一个 message 对象,role 设置为"user",内容在 parts 中。
    • Helloworld 智能体的 execute 方法将排队一个单一的"Hello World"消息。DefaultRequestHandler 将检索此消息并将其作为响应发送。
    • response 将是一个 SendMessageResponse 对象,其中包含 SendMessageSuccessResponse(以智能体的 Message 作为结果)或 JSONRPCErrorResponse
  3. 处理任务 ID(Helloworld 的说明性注释)

    Helloworld 客户端(test_client.py)不会直接尝试 get_taskcancel_task,因为简单的 Helloworld 智能体的 execute 方法,当通过 message/send 调用时,会导致 DefaultRequestHandler 返回直接的 Message 响应而不是 Task 对象。更复杂的智能体(如 LangGraph 示例)会从 message/send 返回 Task 对象,其 id 可用于 get_taskcancel_task

  4. 发送流式消息(send_message_streaming

    
    
    • 此方法调用智能体的 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}}

(实际 ID 如 xxxxxxxxyyyyyyyyzzzzzzzzwwwwwwww 将是不同的 UUID/请求 ID)

这确认您的服务器正在使用更新的 SDK 结构正确处理基本的 A2A 交互!

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