Building an A2A Client in Python: A Hands-On Guide

Most A2A content stops at the Agent Card: how to publish one, how to sign it, how a client resolves it. That is the easy half. The half that decides whether the integration works is the client - the code that discovers a remote agent, opens a conversation, streams its progress, picks the conversation back up after the agent asks a clarifying question, and gets notified when a task finishes while you were not looking.

This post is the code-along half. We build a working A2A client in Python against the official SDK (pip install a2a-sdk), covering the full client surface: card resolution, tasks/send and tasks/sendSubscribe, polling with tasks/get, multi-turn continuation on input-required, and push notifications for disconnected scenarios. We use the current protocol shape from the v1.0 line: JSON-RPC 2.0 over HTTP(S), Agent Cards at /.well-known/agent-card.json, and Server-Sent Events for streaming.

By the end you will have a minimal-but-complete client you can point at any A2A server - the reference Helloworld agent, a LangGraph agent, or your own - plus the error-handling habits that keep it alive when the network does not cooperate.

The shape of the protocol

Three facts about A2A drive every design decision in the client:

• JSON-RPC 2.0 over HTTP(S). Every operation - discovery, sending messages, reading task state, configuring push notifications - is a JSON-RPC method on an HTTP endpoint.
• The Agent Card is the contract. Before doing anything else, a client fetches the card, a self-describing JSON manifest declaring identity, supported interfaces, capabilities (streaming, push notifications, extended card), skills, and security requirements. Everything the client configures comes from the card, not from out-of-band knowledge.
• Tasks, not request/response. A message starts or continues a task - a durable object with a state machine (submitted, working, input-required, completed, failed, canceled, rejected), artifacts, and history. The client’s job is to drive that state machine, not to read a single response body.

The official SDKs are listed in the A2A repository: Python (pip install a2a-sdk), JavaScript, Java, .NET, Go, and Rust. We use the Python one.

Step 1: Resolve the Agent Card

Start every client the same way - fetch the card and let it configure you:

import httpx

from a2a.client import A2ACardResolver


async def resolve_card(base_url: str) -> object:
    async with httpx.AsyncClient() as httpx_client:
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url=base_url,
            # agent_card_path=''  # set if the server publishes elsewhere
        )
        return await resolver.get_agent_card()

A2ACardResolver is a convenience wrapper: get_agent_card() fetches the AgentCard from the server’s /.well-known/agent-card.json endpoint (or a path you override) and returns it. Inspect the card before building anything else - it is your configuration source:

card = await resolve_card('http://127.0.0.1:9999')
print(card.name, card.version)
print('streaming:', card.capabilities.streaming)
print('push:', card.capabilities.pushNotifications)
for iface in card.interfaces:
    print('interface:', iface.url, iface.protocol)

Two capability flags gate the rest of this post. capabilities.streaming: true is the server declaring it supports SendStreamingMessage over SSE; capabilities.pushNotifications: true declares webhook support. A client that ignores these flags and unconditionally opens streams will get HTTP errors against honest servers - always branch on the card.

An A2A server may also publish an extended card behind authentication, containing skills or interfaces the public card omits. The reference Helloworld sample does exactly this: the public card advertises a hello_world skill, while the extended card (fetched with credentials) adds a super_hello_world skill. Treat the public card as the discovery minimum and the extended card as the authenticated contract.

Step 2: Build the client from the card

With the card in hand, create a client. ClientConfig selects the transport mode, and create_client assembles a client configured from the card’s interfaces:

from a2a.client import ClientConfig, create_client
from a2a.helpers import new_text_message
from a2a.types import Role, SendMessageRequest


async def build_client(card, streaming: bool = False):
    config = ClientConfig(streaming=streaming)
    return await create_client(agent=card, client_config=config)

Note the split: streaming=False gives you the plain request/response client, streaming=True a streaming one that speaks SSE. The tutorial client builds two and uses each for what it is good at - a non-streaming client for simple questions, a streaming client when you want incremental updates.

Step 3: Send a message and read the Task

Construct a message, wrap it in a request, and send it. client.send_message returns an async iterator that yields the server’s response - for a non-streaming client, a single final chunk containing the Task (or a Message):

async def ask_once(base_url: str, text: str):
    card = await resolve_card(base_url)
    client = await build_client(card, streaming=False)
    try:
        message = new_text_message(text, role=Role.ROLE_USER)
        request = SendMessageRequest(message=message)
        async for chunk in client.send_message(request):
            return chunk
    finally:
        await client.close()

The returned Task is the durable object - read its status rather than guessing from prose:

task = await ask_once('http://127.0.0.1:9999', 'Say hello.')
print(task.id, task.status.state)
for artifact in task.artifacts or []:
    for part in artifact.parts:
        print('artifact:', part.text)

In the reference Helloworld run, this yields a task whose status is TASK_STATE_COMPLETED with a single artifact containing “Hello, World!”. The task’s id and its context_id are the two handles you will thread through everything that follows: id addresses this task, context_id groups the conversation it belongs to. Every subsequent turn that should continue the same conversation must carry both.

For polling an existing task - after a process restart, or when the agent runs longer than you want to hold a connection - the pattern is the same client, tasks/get via the SDK’s task query with the task id:

# Pseudocode for the SDK surface; the exact helper name is version-specific
task = await get_task(client, task_id=task.id)
while task.status.state in ('TASK_STATE_SUBMITTED', 'TASK_STATE_WORKING'):
    await asyncio.sleep(2)
    task = await get_task(client, task_id=task.id)

Polling is always available; streaming and push are optimizations the card may or may not offer.

Step 4: Stream updates over SSE

When the card declares capabilities.streaming: true, switch the client to streaming and send the same request. The server responds 200 OK with Content-Type: text/event-stream, and each SSE event carries a JSON-RPC response whose result is one of three things:

• The Task itself (initial state),
• A TaskStatusUpdateEvent - a lifecycle transition (working, input-required, completed) plus any intermediate agent message, or
• A TaskArtifactUpdateEvent - a new or updated artifact, with append and lastChunk fields for reassembling streamed chunks.

async def ask_streaming(base_url: str, text: str):
    card = await resolve_card(base_url)
    client = await build_client(card, streaming=True)
    try:
        message = new_text_message(text, role=Role.ROLE_USER)
        request = SendMessageRequest(message=message)
        events = []
        async for chunk in client.send_message(request):
            events.append(chunk)
            kind = type(chunk).__name__
            print('event:', kind)
        return events
    finally:
        await client.close()

Run this against the Helloworld server and you see the canonical four-chunk sequence: the initial task (SUBMITTED), a status_update for WORKING with the agent’s “Processing request…” message, an artifact_update carrying the result, and a final status_update for COMPLETED.

Two streaming rules from the spec that save debugging time later. First, the server closes the stream when the task reaches a terminal or interrupted state - COMPLETED, FAILED, CANCELED, REJECTED, or INPUT_REQUIRED - and sends no further updates; your async for loop ending is the signal. Second, a broken connection is resumable: if the SSE stream drops while the task is still active, the client can reattach to the task’s stream with the SubscribeToTask RPC method rather than restarting the work.

Step 5: Multi-turn on input-required

The interesting A2A behavior is not a one-shot completed task - it is an agent that stops mid-task to ask a question. The LangGraph currency agent in the reference samples does this: ask “how much is 100 USD?” and the agent cannot proceed, so it enqueues a TaskStatusUpdateEvent with state = TASK_STATE_INPUT_REQUIRED and a clarifying question as the status message. The stream closes after that event.

The client’s move is to continue the same task, sending the missing information with the original taskId and contextId:

async def continue_task(base_url: str, task, follow_up: str):
    card = await resolve_card(base_url)
    client = await build_client(card, streaming=True)
    try:
        message = new_text_message(
            follow_up, role=Role.ROLE_USER, task_id=task.id, context_id=task.context_id
        )
        request = SendMessageRequest(message=message)
        async for chunk in client.send_message(request):
            yield chunk
    finally:
        await client.close()

Thread task.id and task.context_id into the follow-up message, and the server resumes the existing task with its full history instead of starting over. Omit them and you have opened a brand-new conversation - the single most common integration bug in A2A clients. A robust client therefore treats INPUT_REQUIRED not as an error but as a normal branch: read task.status.message for the agent’s question, present it to the user, and on their reply re-enter the streaming path with both handles.

Step 6: Push notifications for long tasks

Holding a connection open for a task that runs minutes or days is wasteful, and some clients - mobile apps, serverless functions - cannot hold connections at all. When the card declares capabilities.pushNotifications: true, the client supplies a TaskPushNotificationConfig: an HTTPS webhook url, an optional token for client-side validation, and optional authentication details for how the A2A server should authenticate to your webhook. You can attach it in the initial SendMessage/SendStreamingMessage request, or to an existing task with CreateTaskPushNotificationConfig.

The flow: the server POSTs a StreamResponse payload (a task, message, statusUpdate, or artifactUpdate object) to your webhook on significant state changes - terminal states, input-required, auth-required. Your webhook then uses GetTask with the task id from the notification to fetch the full updated task.

Do not skip the security section of the spec here, because push is server-initiated outbound HTTP - the exact shape SSRF exploits are made of:

• Servers must not blindly POST to any client-supplied URL. The spec’s mitigations: domain allowlists, ownership verification, and egress controls.
• Servers MUST authenticate to the webhook per the configured scheme - bearer tokens (OAuth 2.0), API keys, HMAC signatures, or mutual TLS.
• Webhook receivers MUST verify the notification really came from the A2A server, reject stale timestamps (notifications SHOULD carry a timestamp; reject ones too old), and guard against replays with nonces or JWT jti claims. The spec’s reference flow is asymmetric: the server signs a JWT (claims including iss, aud, iat, exp, jti, and the task id) with a private key and publishes public keys at a JWKS endpoint; the webhook resolves the key by the JWT header’s kid, verifies the signature, and validates the claims. Cache JWKS keys.

A client-side Push Notification Service sits at that webhook URL, authenticates the incoming POST, and relays the meaningful part to your application logic.

Error shapes and edge cases

A production A2A client handles these explicitly:

Situation	What you see	Client response
Task completed	Terminal state, artifacts populated	Render artifacts, stop
Agent needs input	INPUT_REQUIRED status, stream closed	Show status.message, continue task with taskId + contextId
Task failed or rejected	FAILED / REJECTED terminal state	Surface the status message; do not blindly retry the same message
SSE drops mid-task	Connection error, task still active	Reattach with SubscribeToTask; never restart the task blindly
Server lacks streaming	Card says capabilities.streaming: false	Use the non-streaming client and poll tasks/get
Auth required	HTTP 401/401-family, or extended-card fetch rejected	Fetch credentials per the card’s security scheme, retry
JSON-RPC error	error object in the response	Log code/message; treat protocol errors as bugs, task failures as business outcomes

The distinction that matters most: a protocol error (malformed response, unknown method) means your client and the server disagree about the contract - fail loudly. A task-level failure (TASK_STATE_FAILED) is the agent doing its job and reporting an outcome - handle it as data.

A minimal complete client

Putting the pieces together:

import asyncio

import httpx

from a2a.client import A2ACardResolver, ClientConfig, create_client
from a2a.helpers import new_text_message
from a2a.types import Role, SendMessageRequest


class MinimalA2AClient:
    def __init__(self, base_url: str):
        self.base_url = base_url.rstrip('/')
        self._http = httpx.AsyncClient()

    async def connect(self):
        resolver = A2ACardResolver(httpx_client=self._http, base_url=self.base_url)
        self.card = await resolver.get_agent_card()
        config = ClientConfig(streaming=bool(self.card.capabilities.streaming))
        self.client = await create_client(agent=self.card, client_config=config)

    async def send(self, text: str, task_id: str | None = None,
                   context_id: str | None = None):
        message = new_text_message(
            text, role=Role.ROLE_USER, task_id=task_id, context_id=context_id
        )
        chunks = []
        async for chunk in self.client.send_message(SendMessageRequest(message=message)):
            chunks.append(chunk)
        return chunks

    async def close(self):
        await self.client.close()
        await self._http.aclose()


async def main():
    c = MinimalA2AClient('http://127.0.0.1:9999')
    await c.connect()
    print('agent:', c.card.name, '| streaming:', c.card.capabilities.streaming)
    chunks = await c.send('Say hello.')
    for chunk in chunks:
        print(type(chunk).__name__, getattr(chunk, 'status', None))
    await c.close()


if __name__ == '__main__':
    asyncio.run(main())

It resolves the card, adapts its transport mode to the declared capability, sends, and cleans up. Everything beyond this - resumable streams, push notification config, extended-card auth - bolts onto these five methods without changing the shape.

Checklist

• Always resolve the Agent Card first and configure the client from it - never hardcode endpoints or capabilities.
• Branch on capabilities.streaming and capabilities.pushNotifications; do not assume either.
• Treat the Task as the durable object: read status.state and artifacts, do not parse prose for outcomes.
• Thread taskId and contextId into every follow-up message, or you silently start a new conversation.
• Treat INPUT_REQUIRED as a normal branch: surface status.message, await the user, continue the same task.
• On SSE drops mid-task, reattach with SubscribeToTask instead of re-sending the original message.
• For push notifications, implement the spec’s security requirements on both ends - webhook URL validation, server authentication, JWT/JWKS verification, replay protection.
• Distinguish protocol errors (fail loudly, it is a contract bug) from task failures (handle as data).

Sources: A2A Protocol repository (a2aproject/A2A), A2A Protocol Documentation, Python Quickstart - Interacting with the Server, Python Quickstart - Streaming and Multi-Turn, A2A Streaming and Asynchronous Operations