7. Streaming and Multi-Turn Interactions (LangGraph Example)

The Hello World example demonstrates the basic mechanics of A2A. For more advanced capabilities, such as robust streaming, task state management, and multi-turn conversations powered by an LLM, turn to the LangGraph example in a2a-samples/samples/python/agents/langgraph/ .

This example features a “Currency Agent” that uses the Gemini model through LangChain and LangGraph to answer currency conversion questions.

Setting up the LangGraph example

1. Get an API key. Create a Gemini API Key if you do not already have one.

2. Set the environment variable. Create a .env file in the a2a-samples/samples/python/agents/langgraph/ directory:

echo "GOOGLE_API_KEY=YOUR_API_KEY_HERE" > .env

Replace YOUR_API_KEY_HERE with your own Gemini API key. Do not commit a real key to a code repository.

3. Install dependencies. The langgraph example has its own pyproject.toml, which declares dependencies such as langchain-google-genai and langgraph. If you hit import errors at runtime, install the missing dependencies listed in that file.

Running the LangGraph server

In your terminal, go to the a2a-samples/samples/python/agents/langgraph/app directory and make sure your virtual environment is activated.

Start the LangGraph agent server:

python __main__.py

The server usually listens on http://localhost:10000.

Interacting with the LangGraph agent

Open a new terminal window, activate your virtual environment, go to a2a-samples/samples/python/agents/langgraph/app, and run its test client:

python test_client.py

When you are done, press Ctrl+C in the terminal running __main__.py to shut down the server.

Key features demonstrated

The langgraph example showcases several important A2A concepts.

1. LLM integration

• agent.py defines CurrencyAgent. It uses ChatGoogleGenerativeAI and LangGraph’s create_react_agent to process user queries.
• This demonstrates how a real LLM can power the agent’s logic.

2. Task state management

• samples/python/agents/langgraph/app/__main__.py initializes a DefaultRequestHandler with an InMemoryTaskStore.
• When the DefaultRequestHandler calls the execute method of CurrencyAgentExecutor, it interacts with the RequestContext, which contains the current task (if any).
• For SendMessage, the DefaultRequestHandler uses the TaskStore to persist and retrieve task state across interactions. The response to SendMessage is a full Task object if the agent’s execution flow involves multiple steps or results in a persistent task.
• run_single_turn_test in test_client.py shows how to get a Task object back and then query it with get_task.

3. Streaming with TaskStatusUpdateEvent and TaskArtifactUpdateEvent

• The execute method of CurrencyAgentExecutor handles both non-streaming and streaming requests, orchestrated by the DefaultRequestHandler.
• As the LangGraph agent processes the request (which might involve calling tools like get_exchange_rate), CurrencyAgentExecutor enqueues different types of events onto the EventQueue:
  • TaskStatusUpdateEvent: for intermediate updates (for example, “Looking up exchange rates…” and “Processing the exchange rates…”).
  • TaskArtifactUpdateEvent: when the final answer is ready, it is enqueued as an artifact, with the lastChunk flag set to True.
  • A final TaskStatusUpdateEvent with state=TaskState.TASK_STATE_COMPLETED signals the end of the task and closes the stream.
• run_streaming_test in test_client.py prints these individual event chunks as they arrive from the server.

4. Multi-turn conversation (TaskState.TASK_STATE_INPUT_REQUIRED)

• The CurrencyAgent can ask for clarification if a query is ambiguous (for example, the user asks “how much is 100 USD?”).
• When this happens, CurrencyAgentExecutor enqueues a TaskStatusUpdateEvent whose status.state is TaskState.TASK_STATE_INPUT_REQUIRED and whose status.message holds the agent’s question (for example, “To which currency would you like to convert?”). The stream closes after this event.
• run_multi_turn_test in test_client.py demonstrates this:
  • It sends an initial ambiguous query.
  • The agent responds (through the DefaultRequestHandler processing the enqueued events) with a Task whose status is TASK_STATE_INPUT_REQUIRED.
  • The client then sends a second message, including the taskId and contextId from the first turn’s Task response, to provide the missing information (“in GBP”). This continues the same task.

For what these states mean, see Life of a Task . For the details of streaming and push notifications, see Streaming and Asynchronous Operations .

Exploring the code

Take some time to read through these files:

• __main__.py: server setup with DefaultRequestHandler and a Starlette application. Note that the AgentCard definition includes capabilities.streaming=True.
• agent.py: the CurrencyAgent with LangGraph, the LLM model, and tool definitions.
• agent_executor.py: CurrencyAgentExecutor, which implements execute (and cancel). It uses the RequestContext to understand the ongoing task and the EventQueue to send back events (TaskStatusUpdateEvent, TaskArtifactUpdateEvent, and a new Task object created implicitly by the first event when no task exists).
• test_client.py: demonstrates various interaction patterns, including retrieving task IDs and using them for multi-turn conversations.

This example gives a much richer illustration of how A2A supports complex, stateful, and asynchronous interactions between agents.

Next step

Continue with Next Steps .

Source: A2A official documentation · Streaming & Multi-Turn