Get started with Agent Mesh

Create your first resumable event stream with A2A or FastAPI.

Choose one browser demo: reconnect to Agent2Agent (A2A) task updates, or resume application-defined server-sent events (SSE) with FastAPI. Neither example needs a model account.

Before you begin

You need Python 3.11 or later, pip, a browser, and a Redis database running version 8.6.2 or later with native producer idempotency (XADD IDMP). The A2A example also requires JavaScript Object Notation (JSON) support. Official Redis distributions with JSON enabled provide it; a source build needs the JSON module built and loaded. The demo application runs locally on port 8000, whichever database option you choose.

Choose your Redis database

Choose Redis Cloud or a local instance, then set REDIS_URL in the terminal where you will run the demo application. Both examples use this connection value.

Continue in the application terminal so that the demo can read REDIS_URL from its environment. If you open a new terminal to try the other example, set REDIS_URL there too.

Resume A2A task updates

A simulated agent produces twenty numbered updates, one per second. Disconnect the browser's subscription while the agent continues, then reconnect to receive the updates you missed.

Install

In your application terminal, create an environment for the A2A example:

mkdir agent-mesh-a2a
cd agent-mesh-a2a
python3 -m venv .venv
source .venv/bin/activate
python -m pip install 'redis-iris[a2a]' uvicorn

Start the demo

1. Save both example files in your agent-mesh-a2a directory:

   • Download a2a-demo.py
   • Download a2a-demo.html

2. Run the application:

   python a2a-demo.py
   

3. Open http://127.0.0.1:8000 in your browser.

This loopback-only example uses one fixed demo owner and needs no token or sign-in. In a deployed application, derive ownership from verified identity and authorize each task subscription.

Disconnect and catch up

1. Select Start task. Numbered updates appear in the page.
2. After a few updates arrive, select Disconnect. The displayed update count stops.
3. Wait a few seconds, then select Reconnect. The missed updates arrive, followed by new updates as the agent continues.
4. Wait for Complete — all 20 updates received.

The task keeps running while the subscription is disconnected. The page remembers the last applied event cursor and reconnects with ?after= to receive retained updates after that cursor. Expand What is being resumed? to see the task ID and saved cursor.

The page uses the browser's built-in EventSource client. It does not restart the task on reconnect. Replay requires retained Redis history; it does not restart interrupted agent execution.

Connect the A2A handler to Redis

The downloaded application defines CountingAgent, its streaming agent card, and the Redis client. These are the integration points:

from redis_iris.agent_mesh.a2a import RedisTaskStore
from redis_iris.agent_mesh.a2a.routes import create_resumable_rest_routes

store = RedisTaskStore(
    redis,
    namespace="quickstart:a2a-browser",
    owner_resolver=lambda context: "local-demo-user",  # Loopback demo only.
)
events = store.event_stream()
handler = store.request_handler(CountingAgent(), agent_card=card, event_stream=events)
routes = create_resumable_rest_routes(
    handler, agent_card=card, store=store, event_stream=events
)

RedisTaskStore saves the task and its updates together. Its paired event stream reads those updates from Redis. create_resumable_rest_routes() exposes subscriptions with event cursors. The agent card enables streaming and advertises resumable_streams_extension().

On reconnect, the browser uses the same task ID and the last cursor it applied:

new EventSource(`/tasks/${taskId}:subscribe?after=${encodeURIComponent(cursor)}`);

The request resumes event delivery for the existing task; it does not submit the work again.

Stop the application with Control-C before trying the FastAPI example on the same port.

Resume FastAPI events

A simulated travel agent plans a day in Rome. It publishes twenty progress events, one per second, then returns a sample itinerary. Disconnect while it works, then reconnect to catch up and receive the answer. The agent uses scripted steps, so you need no model account or live travel service. These are application-defined events, independent of A2A tasks.

Install

In your application terminal, create an environment for the FastAPI example:

mkdir agent-mesh-sse
cd agent-mesh-sse
python3 -m venv .venv
source .venv/bin/activate
python -m pip install 'redis-iris[fastapi-sse]' uvicorn

Start the demo

1. Save both example files in your agent-mesh-sse directory:

   • Download fastapi-sse-demo.py
   • Download fastapi-sse-demo.html

2. Run the application:

   python fastapi-sse-demo.py
   

3. Open http://127.0.0.1:8000 in your browser.

This loopback-only example uses one fixed demo owner and needs no token or sign-in. In a deployed application, authenticate callers and authorize access to each job's events.

Disconnect and catch up

1. Select Start agent. Progress messages show the agent preparing its itinerary.
2. After a few events arrive, select Disconnect. The displayed event count stops.
3. Wait a few seconds, then select Reconnect. The missed events arrive, followed by new events as the agent continues.
4. Wait for Complete — all 20 events received and the agent’s sample itinerary.

The page saves its event cursor and reconnects to the same agent run’s event stream. Expand What is being resumed? to see the run’s job ID and saved cursor. Your application runs the agent independently of the subscription; reconnecting does not start it again.

Publish and serve resumable events

The downloaded application uses the FastAPI SDK's standard setup. It creates its Redis pools from REDIS_URL and manages in-process producer tasks:

from redis_iris.agent_mesh.fastapi_sse import FastAPIRedis

FastAPIRedis(app).lifespan().streams()

Give each demo job its own scope and run ID. Start the producer once, then publish through the writer it receives:

async def produce(writer):
    await writer.append("progress", {"number": number, "message": message})
    await writer.finish("completed", {"count": 20, "answer": AGENT_ANSWER})

await streams.start(produce, scope=f"job:{job_id}", stream_id=job_id)

The excerpt shows the API shape; the downloadable producer loops over all twenty scripted steps. Reconnecting does not call start() again.

The event endpoint serves the authorized job's scope and run:

from redis_iris.agent_mesh.fastapi_sse import StreamBackendDep

@app.get("/jobs/{job_id}/events", response_class=StreamingResponse)
async def events(job_id: UUID, request: Request, streams: StreamBackendDep):
    job = get_job(job_id)
    return await streams.sse(str(job_id), request=request, scope=job["scope"])

The browser reconnects with the last applied cursor. The demo sets REDIS_STREAM_DONE_TTL=3600 unless you override it, retaining a completed run for up to one hour, bounded by the run's original lifetime (one hour by default). Completion never extends that lifetime. Replay requires retained history; it does not restart interrupted execution.

When you're done

Stop the demo application with Control-C. If you started local Redis for this example, stop it with Control-C too.

Stopping the app leaves a Redis Cloud database running. If you created a disposable database and no longer need it, see Delete a database.

Further reading

To add these integrations to your application, follow the A2A SDK guide or FastAPI SSE SDK guide. See the Agent Mesh limitations for beta qualification.