6. Interacting with the Server

With the Helloworld A2A server running, let’s send some requests to it.

The Helloworld test client

The test_client.py script demonstrates how to:

  1. Fetch the Agent Card from the server.
  2. Create a client with create_client.
  3. Send both non-streaming (SendMessage) and streaming (SendStreamingMessage) requests.

Open a new terminal window, activate your virtual environment (do this in the same directory where you created it), and go to the a2a-samples directory:

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

Run the test client:

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

The script’s entry point main() starts an interactive session with the prompt user >: type a sentence to send a non-streaming request to the server and print the response, or type exit to quit. The test_client_workflow function in the same file is a pytest test that walks through four steps: fetching the Agent Card, a non-streaming call, a streaming call, and fetching the extended Agent Card. Its fixture starts the server automatically.

Understanding the client code

Look at the key parts of test_client.py.

1. Fetching the 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 is a convenience class. When get_agent_card() is called, it fetches the AgentCard from the server’s /.well-known/agent-card.json endpoint (based on the provided base URL), which is then used to initialize the client.

2. Initializing the client and sending a non-streaming message

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)
  • The create_client function creates a Client based on the information provided by the AgentCard and a ClientConfig.
  • A Message is built with the new_text_message helper (passing role=Role.ROLE_USER), then wrapped in a SendMessageRequest.
  • The client’s send_message method returns an async iterator that yields the agent’s final Task or Message response. In this example, it is a Task.

3. Initializing the client and sending a streaming message

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)
  • A separate streaming client is created with create_client by setting streaming=True in its ClientConfig.
  • send_message is called again; this time it streams events, and each iteration of the loop prints a chunk as it arrives over the network.
  • Call await client.close() after the loop to release the underlying HTTP connection.

Expected output

Based on the Helloworld executor logic, you will see:

  • The public Agent Card: a formatted summary of the Hello World Agent’s basic information, interfaces, capabilities, I/O modes, and skills.
  • The non-streaming response: a single Task printed in protobuf text format. Its status is TASK_STATE_COMPLETED, it contains an artifact with the greeting text, and its history holds the user message and the agent’s intermediate status message “Processing request…”.
  • The streaming response: chunks arriving in order: the initial task (state TASK_STATE_SUBMITTED), a status_update with TASK_STATE_WORKING, an artifact_update with the result, and a final status_update with TASK_STATE_COMPLETED.
  • The extended Agent Card: the public card plus the additional echo_bot_super_mode skill (fetched through GetExtendedAgentCard).

The id fields in the output differ on every run. Seeing these results confirms that the server handles basic A2A interactions correctly. When you are done, press Ctrl+C in the terminal running __main__.py to shut down the server.

Next step

Helloworld shows the basic mechanics of A2A. Next, explore streaming and multi-turn interactions with the LangGraph example.

Source: A2A official documentation · Interacting with the Server