Integrate Agent Mesh with A2A

Add resumable task updates to an A2A Python application.

Use the Agent Mesh software development kit (SDK) to let clients resume Agent2Agent (A2A) task updates after a disconnect, including through another serving replica. Redis retains the updates and a current Task snapshot. Your application runs the agent and controls access.

For a runnable browser example, start with the A2A quickstart. This guide explains how to add the integration to an existing application.

Install the integration

You need Python 3.11 or later, Redis 8.6.2 or later with native producer idempotency (XADD IDMP) and JavaScript Object Notation (JSON) support, and the official A2A Python SDK 1.2.x.

pip install 'redis-iris[a2a]'

To use both integrations in one application, install redis-iris[a2a,fastapi-sse]. Their shared redis-py range is >=8.0.1,<9; separate Python environments are not required.

Use the quickstart's database setup for a local or Redis Cloud connection. Keep database credentials in application configuration.

Connect your request handler

The application factory in this example accepts your existing AgentExecutor, AgentCard, and Starlette authentication backend. The authentication backend must verify credentials and populate request.user. Set your card's streaming capability to True before constructing the application.

import os
from contextlib import asynccontextmanager

from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from redis.asyncio import Redis
from redis_iris.agent_mesh.a2a import RedisTaskEventStream, RedisTaskStore
from redis_iris.agent_mesh.a2a.routes import (
    create_resumable_rest_routes,
    resumable_streams_extension,
)
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware


def create_app(agent_executor, agent_card, auth_backend):
    commands = Redis.from_url(
        os.environ["REDIS_URL"], max_connections=16, socket_timeout=5
    )
    readers = Redis.from_url(
        os.environ["REDIS_URL"], max_connections=64, socket_timeout=5
    )
    store = RedisTaskStore(commands, namespace="myapp:a2a")
    event_stream = RedisTaskEventStream(readers, store=store)
    handler = store.request_handler(
        agent_executor, agent_card=agent_card, event_stream=event_stream
    )
    agent_card.capabilities.extensions.append(resumable_streams_extension())

    @asynccontextmanager
    async def lifespan(app):
        async with commands, readers:
            try:
                yield
            finally:
                await handler.aclose()

    return Starlette(
        routes=[
            *create_agent_card_routes(agent_card),
            *create_jsonrpc_routes(handler, rpc_url="/"),
            *create_resumable_rest_routes(
                handler, agent_card=agent_card, store=store, event_stream=event_stream
            ),
        ],
        middleware=[Middleware(AuthenticationMiddleware, backend=auth_backend)],
        lifespan=lifespan,
    )

Use the same Redis database, namespace, and ownership policy on every replica. Use primary reads for both snapshots and events. Your application owns credentials, Transport Layer Security (TLS), connection pools, timeouts, and shutdown.

Each waiting subscription occupies a Redis connection. A separate reader pool keeps blocked subscriptions from exhausting connections needed for writes and heartbeats. Set its socket timeout above the event stream's block_ms interval, which defaults to at most 4,000 milliseconds and adapts to a shorter client socket timeout.

RedisTaskStore commits the Task snapshot, version, and event together. The paired event stream reads that log; its publish() method does not append another copy. Keep writing through the official request handler and its task store.

Enforce ownership

By default, the store requires an authenticated user. Behind the SDK's Starlette routes, ownership uses the user's display_name. Ensure this value uniquely identifies a principal across your tenants.

For a different identity model, pass owner_resolver to RedisTaskStore. The callable receives the official ServerCallContext and must return a nonempty, stable owner derived from verified identity. Never derive it from request bodies, task IDs, or unverified headers.

The official handler and resumable routes check ownership before subscribing. The internal RedisTaskEventStream reader has no request context: do not expose it directly to callers without that check. Task IDs and event cursors grant no access.

Start and resume a subscription

Pass the handler's same agent_card to create_resumable_rest_routes(); its streaming capability must be enabled. The factory adds event cursors to the HTTP+JSON binding's POST /message:stream and GET or POST /tasks/{id}:subscribe routes. Other REST routes retain the official SDK's behavior.

To follow an existing active task, send an authenticated request to /tasks/<task-id>:subscribe with A2A-Version: 1.0. Without a cursor, the response starts with the current Task snapshot and follows new updates.

Task frames use server-sent events (SSE) and carry an id: containing the Task version. After applying a frame, save its ID. On reconnect, send that value in Last-Event-ID to receive retained updates strictly after it. The routes also accept the after query parameter.

For example, these are the request headers for a reconnect using your application's bearer authentication:

GET /tasks/<task-id>:subscribe HTTP/1.1
Accept: text/event-stream
A2A-Version: 1.0
Authorization: Bearer <application-access-token>
Last-Event-ID: <last-applied-task-version>

Use an SSE client that parses frames and handles reconnect status codes. A fetch-based browser client can supply both the version and authorization headers. Native EventSource cannot set those headers. To use it, you need an explicit fixed-version server policy and browser-compatible authentication, as demonstrated by the local quickstart. The route factory does not install that policy.

A missing version header is interpreted as A2A 0.3 and rejected by the resumable routes. Declaring resumable_streams_extension() in the card advertises support; it does not select the request version. The official A2A client does not send Last-Event-ID, so it receives standard snapshot-then-live subscriptions.

Message-only replies and error frames have no Task-version ID. They must not advance the saved cursor. Task versions are not raw Redis stream entry IDs.

The route factory's credential_expiry callback and max_connection_duration limit also bound the wait for the first frame. If credentials expire before the server sends response headers, the request returns HTTP 401; refresh credentials before reconnecting. If the maximum connection duration expires first, the request returns HTTP 503 with Retry-After: 1; wait one second, then reconnect with your saved cursor.

Handle replay and completion

An update that moves a Task to a terminal, input-required, or auth-required state ends the HTTP subscription. If the Task already requires input or authentication when you subscribe, the stream can remain open. Without a cursor, it sends the waiting Task snapshot and then waits. With a cursor matching the current version, it waits without sending a Task frame.

Act on the Task's state and provide the required input or credentials through your application's A2A interaction. Do not wait for the stream to close. Disconnecting a subscriber does not cancel execution.

For cancellation, use the official handler's cancellation path. A cancellation can supersede a stale version while the stored Task is nonterminal. A later write cannot revive a terminal Task. These guards protect Redis updates; they do not undo external effects.

Monitor executing agents

store.request_handler() applies store.heartbeat(executor, interval=5.0, alive_ttl=15.0) to track the running agent under a process lease. It checks stored terminal state and stops local execution when another replica cancels or finishes the Task. Set alive_ttl greater than interval.

A store-backed subscriber or get() marks a submitted or working Task failed with a producer_lost status message when its producer lease expires. Tasks waiting for input or authentication, and tasks without tracked heartbeats, are excluded. To resume such a Task in another process instead, see Hand off a task to another process.

The wrapper does not restart computation. Without it, an agent on another replica learns of cancellation at its next write. With A2A SDK 1.2.3, continued emission after that cancellation can also leave handler shutdown waiting indefinitely; the wrapper stops the agent first.

Hand off a task to another process

One process can start a Task and another process can finish it. Both write to the same Task and its ordered event log, so subscribers on any replica see one sequence of updates, whichever process produced them.

Before the heartbeat wrapper runs your executor, it claims the Task for its process. Each save carries that claim. After another process claims the Task, the store rejects saves from the previous process with ConcurrentTaskModificationError, so a process that stalls and later wakes up cannot overwrite newer work. That process also stops its executor at its next successful heartbeat.

The wrapper grants a claim when the Task is new, when no process runs it, when it waits for input or authentication, or when the process that ran it stopped sending heartbeats. It refuses a claim for a terminal or deleted Task, a Task the caller doesn't own, and a Task that a process sending heartbeats is running or has just claimed. After a refused claim, the executor doesn't run, the Task keeps its current state, and the message is not added to the Task's history. If your executor's first update on a resumed Task is a status or artifact update, and you don't limit history_length, check the returned Task's history to tell whether your message was accepted.

Continue a waiting task in another process

A Task that waits for input or authentication is not running anywhere. The next message can reach any replica. That replica's wrapper claims the Task and runs your executor with the stored Task as context.current_task. No extra configuration is required. While one replica is processing a reply, other replicas refuse a second reply to the same Task.

Take over a long-running task

By default, a reader that notices a lost heartbeat marks a submitted or working Task failed with producer_lost. This includes the task read that a resume request makes. To let another process resume the Task instead, create the store with takeover=True on every replica:

store = RedisTaskStore(commands, namespace="myapp:a2a", takeover=True)

To resume a Task, send it a message through the official request handler in the process that takes it over. The handler reloads the Task, the wrapper claims it, and your executor runs with the stored Task. A worker can make that call in-process, using a ServerCallContext that carries the Task owner's verified identity:

import uuid

from a2a.types import Message, Part, Role, SendMessageRequest


async def resume(handler, task_id, context_id, owner_context):
    request = SendMessageRequest(
        message=Message(
            message_id=f"resume-{uuid.uuid4()}",
            role=Role.ROLE_USER,
            task_id=task_id,
            context_id=context_id,
            parts=[Part(text="resume")],
        )
    )
    request.configuration.return_immediately = True
    return await handler.on_message_send(request, owner_context)

Build owner_context in trusted worker code from the identity that owns the Task. Never build it from request input. The message becomes part of the Task's history.

Have your executor record its progress in the Task, for example as one artifact per completed step, and skip completed steps when context.current_task already contains them:

from a2a.server.agent_execution import AgentExecutor


class TripPlanner(AgentExecutor):
    async def execute(self, context, event_queue):
        task = context.current_task
        done = {artifact.artifact_id for artifact in task.artifacts} if task else set()
        if "flights" not in done:
            ...  # book flights, then emit the "flights" artifact
        if "hotels" not in done:
            ...  # book hotels, then emit the "hotels" artifact
        ...  # emit the completed status

    async def cancel(self, context, event_queue):
        ...  # emit the canceled status

The integration does not detect stopped processes or schedule resumption. Track the Tasks your workers run, for example in your job queue, and resume or cancel those whose process stopped. With takeover=True, a running Task that nobody resumes stays working until active_ttl expires, and its subscribers stay connected until then. A step that the first process performed but did not record runs again in the next process, so make external effects safe to repeat.

Configure retention and limits

Pass these options to RedisTaskStore:

Heartbeats do not extend Task retention. Set active_ttl longer than the longest interval without a saved update.

The byte limits count serialized protocol buffer bytes; stored JSON payloads use more space. Size the event limit and log length for your memory budget. An oversized payload raises A2ACapacityExceeded without changing the snapshot or log.

Owner metadata prevents another owner from reusing a Task ID until it expires. Deleting an owned Task removes its snapshot and events while retaining ownership metadata for metadata_ttl. After metadata expiry, reuse cannot be detected.

Storage and internal reader calls can raise A2AStorageError, A2ACapacityExceeded, or A2AStreamExpired. Stale or conflicting writes raise the official SDK's ConcurrentTaskModificationError. Let the handler and routes translate these into protocol responses rather than exposing storage exceptions to clients.

Task listing scans an owner's Task index and suits modest inventories per owner. It is not a search index or a consistent snapshot during concurrent writes.

See the beta limits for deployment qualification. Redis durability remains application-owned; replay does not guarantee exactly-once external effects.