From 978a8b5672d6d43c44cd26065bd194cc4c45b283 Mon Sep 17 00:00:00 2001 From: Julien Ruaux Date: Tue, 27 Oct 2026 14:31:19 +0100 Subject: [PATCH 1/9] docs(agent-mesh): add preview overview and browser quickstarts Explain resumable event streams and provide A2A and FastAPI browser demos with Redis Cloud and local connection options. Link the preview from the Iris developer and operator indexes. --- content/develop/ai/context-engine/_index.md | 1 + .../ai/context-engine/agent-mesh/AGENTS.md | 27 ++ .../ai/context-engine/agent-mesh/_index.md | 65 +++++ .../context-engine/agent-mesh/quickstart.md | 230 ++++++++++++++++++ content/operate/iris/_index.md | 4 + static/code/agent-mesh/a2a-demo.html | 179 ++++++++++++++ static/code/agent-mesh/a2a-demo.py | 127 ++++++++++ static/code/agent-mesh/fastapi-sse-demo.html | 177 ++++++++++++++ static/code/agent-mesh/fastapi-sse-demo.py | 131 ++++++++++ 9 files changed, 941 insertions(+) create mode 100644 content/develop/ai/context-engine/agent-mesh/AGENTS.md create mode 100644 content/develop/ai/context-engine/agent-mesh/_index.md create mode 100644 content/develop/ai/context-engine/agent-mesh/quickstart.md create mode 100644 static/code/agent-mesh/a2a-demo.html create mode 100644 static/code/agent-mesh/a2a-demo.py create mode 100644 static/code/agent-mesh/fastapi-sse-demo.html create mode 100644 static/code/agent-mesh/fastapi-sse-demo.py diff --git a/content/develop/ai/context-engine/_index.md b/content/develop/ai/context-engine/_index.md index ef97b98428..b047dd77f7 100644 --- a/content/develop/ai/context-engine/_index.md +++ b/content/develop/ai/context-engine/_index.md @@ -22,6 +22,7 @@ Redis Iris eliminates the infrastructure burden of building context-aware AI age {{< tile-card icon="images/dev/icons/icon-redis-iris-64-duotone.png" title="Context Gateway" description="One MCP endpoint for the tools of any MCP server and your Redis data" url="/develop/ai/context-engine/context-gateway" >}} {{< tile-card icon="images/dev/icons/icon-redis-iris-64-duotone.png" title="Agent Playbooks" description="Publish reusable answers and procedures for agents" url="/develop/ai/context-engine/agent-playbook" >}} {{< tile-card icon="images/dev/icons/icon-redis-iris-64-duotone.png" title="Lens" description="Debug agent answers by tracing the context behind each turn" url="/develop/ai/context-engine/lens" >}} + {{< tile-card icon="images/dev/icons/icon-redis-code-64-duotone.png" title="Agent Mesh (preview)" description="Share A2A task state across Python replicas and resume event delivery" url="/develop/ai/context-engine/agent-mesh" >}} ## What is Redis Iris? diff --git a/content/develop/ai/context-engine/agent-mesh/AGENTS.md b/content/develop/ai/context-engine/agent-mesh/AGENTS.md new file mode 100644 index 0000000000..519e556df0 --- /dev/null +++ b/content/develop/ai/context-engine/agent-mesh/AGENTS.md @@ -0,0 +1,27 @@ +# Agent Mesh documentation + +- Agent Mesh is a preview Python SDK integration distributed through `redis-iris`. + Imports use `redis_iris.agent_mesh`. +- Lead with resumable event streams on Redis. Present A2A task state storage as + a supporting capability, and A2A and FastAPI SSE as two Python integration paths. +- Distinguish the independent `[a2a]` and `[fastapi-sse]` extras. The A2A + integration does not require the FastAPI SSE integration. +- Use Agent2Agent (A2A), server-sent events (SSE), Task snapshots, event replay, + and application-owned execution. Explain acronyms on first use on each page. +- Write for the intended released package. Use package-index installation + commands with the documented extras, not PR refs, source installs or commit pins. + Do not invent a release version. +- Applications own authentication, Redis clients, agent execution and lifecycle. + Task IDs and cursors are not credentials. Derive ownership from verified identity + in deployed applications. The explicitly authorized loopback-only browser demos + use a fixed demo owner without authentication; keep that exception local. +- Replay is bounded by retention and can fall back to a current snapshot. Do not + promise automatic execution restart or exactly-once external effects. +- Keep the standalone Go mesh and its `agentmesh` HTTP client out of this SDK + quickstart. Their deployment and Redis requirements are separate. +- Link detailed implementation and API guidance to redis-iris documentation. + Preserve the product preview banner and qualification limitations. + +- Keep two public pages: an overview and one quickstart with both integrations. + Each example reaches one first result. Keep feature walkthroughs and detailed + configuration outside the quickstart. diff --git a/content/develop/ai/context-engine/agent-mesh/_index.md b/content/develop/ai/context-engine/agent-mesh/_index.md new file mode 100644 index 0000000000..7d13554fff --- /dev/null +++ b/content/develop/ai/context-engine/agent-mesh/_index.md @@ -0,0 +1,65 @@ +--- +alwaysopen: false +categories: +- docs +- develop +- ai +description: Create resumable event streams with Agent Mesh through A2A or FastAPI. +hideListLinks: true +linktitle: Agent Mesh +title: Redis Agent Mesh +weight: 40 +bannerText: Redis Agent Mesh is currently available in preview. Features and behavior are subject to change. +bannerChildren: true +--- + +Redis Agent Mesh provides resumable event streams on top of Redis. Your application keeps producing events when a client disconnects. When the client reconnects, it can receive retained updates after its last event cursor. + +The Agent Mesh software development kit (SDK) offers two Python integration paths: Agent2Agent (A2A) task updates through the official A2A Python SDK, and server-sent events (SSE) through FastAPI. Both are available in the `redis-iris` Python distribution. Your application owns execution, authentication, Redis connections, and resource lifecycle. + +
+ {{< tile-card icon="images/dev/icons/icon-redis-code-64-duotone.png" title="Quickstart" description="Create your first resumable event stream with A2A or FastAPI" url="/develop/ai/context-engine/agent-mesh/quickstart" >}} +
+ +## Why use resumable event streams? + +An agent response can take seconds or minutes as it calls tools, waits for results, and prepares an answer. If the user's connection drops halfway through, they need a way to catch up and receive the result without submitting the request again. + +Resumable streams let you: + +- **Show progress throughout the work.** Deliver progress updates, tool results, and completed output as your application produces them. +- **Reconnect to the same work.** The application can keep publishing while the client is disconnected. The client returns with its saved cursor to receive retained updates it missed and follow new ones. +- **Resume through another serving replica.** Redis gives replicas a shared event log. Replicas with the same Redis configuration and access policy can serve a reconnecting client, independently of which process produced the events. + +Resumption recovers event delivery. Your application remains responsible for restarting work if the process executing it stops. + +## Choose an integration + +| Integration | Use it to | Python module | Install with | +| --- | --- | --- | --- | +| A2A server | Resume retained task updates after reconnecting, including through another server replica | `redis_iris.agent_mesh.a2a` | `pip install 'redis-iris[a2a]'` | +| FastAPI event streaming | Publish an event log and replay retained server-sent events (SSE) after reconnecting | `redis_iris.agent_mesh.fastapi_sse` | `pip install 'redis-iris[fastapi-sse]'` | + +These integrations are independent. Install the integration you use. + +## How it works + +Both integrations retain events in Redis and give clients cursors for resuming delivery. Clients reconnect with their last event cursor to receive retained updates. + +For A2A, connect `RedisTaskStore` and its paired event stream to the official SDK's request handler, and expose the resumable routes. The integration also stores Task snapshots alongside their updates. Replicas using the same database, namespace, and ownership policy can serve the subscriptions. + +For FastAPI, publish application events with `StreamBackend` and expose a streaming endpoint with `sse_response()`. Event production is independent of client connections. + +## Requirements and limits + +- Both integrations require Python 3.11 or later and Redis 8.6 or later with native producer idempotency (`XADD IDMP`). The A2A integration targets the official A2A Python SDK 1.2.x. +- Derive ownership from verified identity. Task IDs and cursors grant no access; authorize requests and reconnections in your application. +- Retention bounds replay. A2A can fall back to the current snapshot when event history is unavailable. If events are no longer retained, a FastAPI application must provide another way for clients to retrieve the current state. +- Your application owns execution and Redis durability. Stored tasks do not restart interrupted computation or guarantee exactly-once external effects. +- Standalone Redis has local validation evidence. Cluster, Enterprise, failover, and sustained capacity remain unqualified for this preview. + +## Get started + +Follow the [quickstart](/content/develop/ai/context-engine/agent-mesh/quickstart.md) to [resume A2A task updates](/content/develop/ai/context-engine/agent-mesh/quickstart.md#resume-a2a-task-updates) or [resume FastAPI events](/content/develop/ai/context-engine/agent-mesh/quickstart.md#resume-fastapi-events). + +For detailed configuration and behavior, see the [A2A hosting guide](https://github.com/redis/redis-iris/blob/main/docs/user_guides/a2a-server.md) and [FastAPI integration guide](https://github.com/redis/redis-iris/blob/main/docs/user_guides/resumable-sse.md). diff --git a/content/develop/ai/context-engine/agent-mesh/quickstart.md b/content/develop/ai/context-engine/agent-mesh/quickstart.md new file mode 100644 index 0000000000..178fe7cc0b --- /dev/null +++ b/content/develop/ai/context-engine/agent-mesh/quickstart.md @@ -0,0 +1,230 @@ +--- +alwaysopen: false +categories: +- docs +- develop +- ai +description: Create your first resumable event stream with A2A or FastAPI. +hideListLinks: true +linktitle: Quickstart +title: Get started with Agent Mesh +weight: 10 +--- + +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 or later with native producer idempotency (`XADD IDMP`). 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. + +{{< multitabs id="agent-mesh-redis" tab1="Redis Cloud" tab2="Local Redis" >}} + +1. Use a dedicated Redis Cloud database running Redis 8.6 or later. See [Create a database](/content/operate/rc/databases/create-database/_index.md) and [supported database versions](/content/operate/rc/databases/version-management/_index.md#supported-database-versions). +1. [Enable Transport Layer Security (TLS)](/content/operate/rc/security/database-security/tls-ssl.md#enable-tls) and download the server certificate bundle. Get your database endpoint, username, and password from the [connection guide](/content/operate/rc/databases/connect/_index.md). +1. Set the connection URL in your application terminal: + + ```bash + export REDIS_URL='rediss://:@:/0?ssl_ca_certs=' + ``` + +Replace the placeholders with your database values and the certificate bundle's absolute path. Percent-encode special characters in the username, password, and certificate path; see [connection URLs](/content/develop/clients/connection-urls.md). Keep this URL out of source control and logs. + +These are Redis database credentials. The local browser demo still needs no token or sign-in. Redis Cloud deployment and failure qualification remain separate from the [preview's tested coverage](/content/develop/ai/context-engine/agent-mesh/_index.md#requirements-and-limits). + +-tab-sep- + +You need a local `redis-server` executable, another terminal, and free port 6389. + +1. Start disposable Redis in that other terminal: + + ```bash + redis-server --bind 127.0.0.1 --port 6389 --save '' --appendonly no + ``` + + Leave it running. If the port is in use, stop rather than reusing an existing database. Persistence is disabled for this local example. + +1. Set the connection URL in your application terminal: + + ```bash + export REDIS_URL='redis://127.0.0.1:6389/0' + ``` + +{{< /multitabs >}} + +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: + +```bash +mkdir agent-mesh-a2a +cd agent-mesh-a2a +python3 -m venv .venv +source .venv/bin/activate +python -m pip install 'redis-iris[a2a]' 'a2a-sdk[http-server]>=1.2.3,<1.3' uvicorn +``` + +### Start the demo + +1. Save both example files in your `agent-mesh-a2a` directory: + + - {{< download "Download a2a-demo.py" "../../../../../code/agent-mesh/a2a-demo.py" >}} + - {{< download "Download a2a-demo.html" "../../../../../code/agent-mesh/a2a-demo.html" >}} + +1. Run the application: + + ```bash + python a2a-demo.py + ``` + +1. Open 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. +1. After a few updates arrive, select **Disconnect**. The displayed update count stops. +1. Wait a few seconds, then select **Reconnect**. The missed updates arrive, followed by new updates as the agent continues. +1. 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: + +```python +from a2a.server.request_handlers import DefaultRequestHandler +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 = DefaultRequestHandler( + agent_executor=store.heartbeat(CountingAgent()), + agent_card=card, + task_store=store, + event_stream=events, +) +routes = create_resumable_rest_routes(handler, 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: + +```javascript +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 + +An application job publishes twenty progress events, one per second. Disconnect its subscription while the job continues, then reconnect to receive the events you missed. These are application-defined events, independent of A2A tasks. + +### Install + +In your application terminal, create a separate environment for the FastAPI example: + +```bash +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 "Download fastapi-sse-demo.py" "../../../../../code/agent-mesh/fastapi-sse-demo.py" >}} + - {{< download "Download fastapi-sse-demo.html" "../../../../../code/agent-mesh/fastapi-sse-demo.html" >}} + +1. Run the application: + + ```bash + python fastapi-sse-demo.py + ``` + +1. Open 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 job**. Numbered progress events appear in the page. +1. After a few events arrive, select **Disconnect**. The displayed event count stops. +1. Wait a few seconds, then select **Reconnect**. The missed events arrive, followed by new events as the job continues. +1. Wait for **Complete — all 20 events received**. + +The page saves its event cursor and reconnects to the same job's event stream. Expand **What is being resumed?** to see the job ID and saved cursor. The job continues independently of the subscription; reconnecting does not start it again. + +### Publish and serve resumable events + +The downloaded application owns its Redis clients, startup, and job execution. Its streaming backend uses separate command and relay clients: + +```python +from redis_iris.agent_mesh.fastapi_sse import StreamBackend, StreamSettings + +backend = StreamBackend( + commands, blocking_client=relays, + settings=StreamSettings(prefix="quickstart:sse-browser", stream_bucket_count=2), +) +``` + +Create a log when starting a job, append its progress events, and finish it when work completes: + +```python +await backend.create(stream, ttl=3600) +await backend.append( + stream, event="progress", data={"number": number}, + producer="demo", idempotency_key=f"progress-{number}", +) +await backend.finish( + stream, event="completed", data={"count": 20}, + producer="demo", idempotency_key="completed", +) +``` + +The event endpoint uses `sse_response()` to serve both retained and live events: + +```python +from redis_iris.agent_mesh.fastapi_sse import StreamBackendDep, sse_response + +@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 sse_response(request, backend=streams, stream=job["stream"]) +``` + +The browser's built-in `EventSource` reconnects with the last applied cursor. This demo retains each log for one hour. Replay requires retained history, and event delivery does not restart interrupted job 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](/content/operate/rc/databases/delete-database.md). + +## Further reading + +For configuration and behavior beyond these first examples, see the [A2A hosting guide](https://github.com/redis/redis-iris/blob/main/docs/user_guides/a2a-server.md), [FastAPI integration guide](https://github.com/redis/redis-iris/blob/main/docs/user_guides/resumable-sse.md), and [Agent Mesh limitations](/content/develop/ai/context-engine/agent-mesh/_index.md#requirements-and-limits). diff --git a/content/operate/iris/_index.md b/content/operate/iris/_index.md index 11cb775d07..9a1d537f19 100644 --- a/content/operate/iris/_index.md +++ b/content/operate/iris/_index.md @@ -40,3 +40,7 @@ Deploy Redis Iris services on Kubernetes when you need to operate them on your o - [Self-managed Redis Agent Memory]({{< relref "/operate/iris/agent-memory/self-managed" >}}) - [Self-managed Redis Context Gateway]({{< relref "/operate/iris/context-gateway/self-managed" >}}) + +## Agent Mesh SDK preview + +Agent Mesh provides Python SDK integrations for shared A2A task storage and resumable event delivery. Run it in your application with your own Redis connections. See the [Agent Mesh overview](/content/develop/ai/context-engine/agent-mesh/_index.md) and [quickstart](/content/develop/ai/context-engine/agent-mesh/quickstart.md). diff --git a/static/code/agent-mesh/a2a-demo.html b/static/code/agent-mesh/a2a-demo.html new file mode 100644 index 0000000000..016d84e9eb --- /dev/null +++ b/static/code/agent-mesh/a2a-demo.html @@ -0,0 +1,179 @@ + + + + + + Agent Mesh: reconnect without missing updates + + + +
+
AGENT MESH · LOCAL A2A DEMO
+

Reconnect without missing updates

+

A simulated agent produces 20 updates, one per second. Disconnect your subscription, wait a few seconds, and reconnect. The agent keeps working while you are away.

+
+
Ready to start
+

Start a task, then disconnect after a few updates arrive.

+
+ + + +
+ + + +
    +
    + What is being resumed? +

    Task: —
    Last applied cursor: —

    +

    The page saves each event cursor after applying its update. Reconnect uses ?after= to request the retained events after that cursor. The browser's native EventSource receives the stream.

    +
    +
    +

    Local demonstration only. There is no sign-in: all requests use one fixed demo owner. Closing a subscription does not cancel agent execution. Replay depends on retained Redis history.

    +
    + + + diff --git a/static/code/agent-mesh/a2a-demo.py b/static/code/agent-mesh/a2a-demo.py new file mode 100644 index 0000000000..7152278405 --- /dev/null +++ b/static/code/agent-mesh/a2a-demo.py @@ -0,0 +1,127 @@ +"""Loopback-only A2A subscription demo. Run: python a2a-demo.py.""" + +import asyncio +import os +from contextlib import asynccontextmanager +from pathlib import Path + +import uvicorn +from a2a.server.agent_execution import AgentExecutor +from a2a.server.request_handlers import DefaultRequestHandler +from a2a.types import ( + AgentCapabilities, AgentCard, AgentInterface, AgentSkill, Artifact, Part, Task, + TaskArtifactUpdateEvent, TaskState, TaskStatus, TaskStatusUpdateEvent, +) +from a2a.utils import TransportProtocol +from redis.asyncio import Redis +from redis_iris.agent_mesh.a2a import 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.base import BaseHTTPMiddleware +from starlette.middleware.trustedhost import TrustedHostMiddleware +from starlette.responses import FileResponse, PlainTextResponse +from starlette.routing import Route + + +class LocalRequestsOnly(BaseHTTPMiddleware): + async def dispatch(self, request, call_next): + # Keep this deliberately unauthenticated example on the local machine. + if request.client is None or request.client.host not in {"127.0.0.1", "::1"}: + return PlainTextResponse("This demo accepts loopback clients only", status_code=403) + origin = request.headers.get("origin") + if origin and origin != f"{request.url.scheme}://{request.headers['host']}": + return PlainTextResponse("Cross-origin requests are disabled", status_code=403) + # Native EventSource cannot set A2A-Version. This demo serves A2A 1.0 only. + if "a2a-version" not in request.headers: + request.scope["headers"] = [*request.scope["headers"], (b"a2a-version", b"1.0")] + return await call_next(request) + + +class CountingAgent(AgentExecutor): + async def execute(self, context, event_queue): + await event_queue.enqueue_event(Task( + id=context.task_id, context_id=context.context_id, + status=TaskStatus(state=TaskState.TASK_STATE_WORKING), + )) + for number in range(1, 21): + # Simulated work, independent of the browser's subscription. + await asyncio.sleep(1) + await event_queue.enqueue_event(TaskArtifactUpdateEvent( + task_id=context.task_id, context_id=context.context_id, + artifact=Artifact( + artifact_id=f"update-{number}", + parts=[Part(text=f"Update {number} of 20")], + ), + last_chunk=True, + )) + await self.finish(context, event_queue, TaskState.TASK_STATE_COMPLETED) + + async def cancel(self, context, event_queue): + await self.finish(context, event_queue, TaskState.TASK_STATE_CANCELED) + + async def finish(self, context, event_queue, state): + await event_queue.enqueue_event(TaskStatusUpdateEvent( + task_id=context.task_id, context_id=context.context_id, + status=TaskStatus(state=state), + )) + + +redis_url = os.environ.get("REDIS_URL") +if not redis_url: + raise ValueError("Set REDIS_URL before running this demo") +redis = Redis.from_url(redis_url, socket_timeout=5) +store = RedisTaskStore( + redis, + namespace="quickstart:a2a-browser", + # Explicit single-user demonstration policy. Use verified identity in an application. + owner_resolver=lambda context: "local-demo-user", +) +events = store.event_stream() +card = AgentCard( + name="Counting agent", description="Produces twenty numbered updates", version="1.0", + capabilities=AgentCapabilities(streaming=True, extensions=[resumable_streams_extension()]), + default_input_modes=["text/plain"], default_output_modes=["text/plain"], + skills=[AgentSkill(id="count", name="Count", description="Counts to twenty", tags=["demo"])], + supported_interfaces=[AgentInterface( + protocol_binding=TransportProtocol.HTTP_JSON, + protocol_version="1.0", url="http://127.0.0.1:8000/", + )], +) +handler = DefaultRequestHandler( + agent_executor=store.heartbeat(CountingAgent()), agent_card=card, + task_store=store, event_stream=events, +) + + +@asynccontextmanager +async def lifespan(app): + async with redis: + await redis.ping() + try: + yield + finally: + # Bound shutdown even if someone stops the demo while a task is working. + await asyncio.wait_for(handler.aclose(), timeout=30) + + +async def home(request): + return FileResponse(Path(__file__).with_name("a2a-demo.html")) + + +app = Starlette( + routes=[Route("/", home), *create_resumable_rest_routes( + handler, store=store, event_stream=events, + )], + lifespan=lifespan, + middleware=[ + Middleware(TrustedHostMiddleware, allowed_hosts=["127.0.0.1", "localhost"]), + Middleware(LocalRequestsOnly), + ], +) + +if __name__ == "__main__": + print("Open http://127.0.0.1:8000 in your browser.") + uvicorn.run(app, host="127.0.0.1", port=8000) diff --git a/static/code/agent-mesh/fastapi-sse-demo.html b/static/code/agent-mesh/fastapi-sse-demo.html new file mode 100644 index 0000000000..762b3b386a --- /dev/null +++ b/static/code/agent-mesh/fastapi-sse-demo.html @@ -0,0 +1,177 @@ + + + + + + Agent Mesh: reconnect without missing updates + + + +
    +
    AGENT MESH · LOCAL FASTAPI SSE DEMO
    +

    Reconnect without missing updates

    +

    An application job publishes 20 progress events, one per second. Disconnect your subscription, wait a few seconds, and reconnect. The job keeps working while you are away.

    +
    +
    Ready to start
    +

    Start a job, then disconnect after a few updates arrive.

    +
    + + + +
    + + + +
      +
      + What is being resumed? +

      Job: —
      Last applied cursor: —

      +

      The page saves each event cursor after applying its update. Reconnect uses ?after= to request the retained events after that cursor. The browser's native EventSource receives the stream.

      +
      +
      +

      Local demonstration only. There is no sign-in: all requests use one fixed demo owner. Closing a subscription does not cancel the application job. Replay depends on retained Redis history.

      +
      + + + diff --git a/static/code/agent-mesh/fastapi-sse-demo.py b/static/code/agent-mesh/fastapi-sse-demo.py new file mode 100644 index 0000000000..8cb6d8aecd --- /dev/null +++ b/static/code/agent-mesh/fastapi-sse-demo.py @@ -0,0 +1,131 @@ +"""Loopback-only FastAPI event demo. Run: python fastapi-sse-demo.py.""" + +import asyncio +import os +import logging +from contextlib import asynccontextmanager +from pathlib import Path +from uuid import UUID, uuid4 + +import uvicorn +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import FileResponse, StreamingResponse +from redis.asyncio import Redis +from redis_iris.agent_mesh.fastapi_sse import ( + StreamBackend, StreamBackendDep, StreamSettings, add_redis_streams, sse_response, +) +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.middleware.trustedhost import TrustedHostMiddleware +from starlette.responses import PlainTextResponse + + +class LocalRequestsOnly(BaseHTTPMiddleware): + async def dispatch(self, request, call_next): + if request.client is None or request.client.host not in {"127.0.0.1", "::1"}: + return PlainTextResponse("This demo accepts loopback clients only", status_code=403) + origin = request.headers.get("origin") + if origin and origin != f"{request.url.scheme}://{request.headers['host']}": + return PlainTextResponse("Cross-origin requests are disabled", status_code=403) + return await call_next(request) + + +redis_url = os.environ.get("REDIS_URL") +if not redis_url: + raise ValueError("Set REDIS_URL before running this demo") +commands = Redis.from_url(redis_url, max_connections=8, socket_timeout=5) +relays = Redis.from_url(redis_url, max_connections=2, socket_timeout=5) +backend = StreamBackend( + commands, blocking_client=relays, + settings=StreamSettings(prefix="quickstart:sse-browser", stream_bucket_count=2), +) +# Application-owned demo state and job execution. No durable job scheduler is supplied. +jobs = {} +producers = set() + + +async def produce(job_id, stream): + try: + for number in range(1, 21): + await asyncio.sleep(1) # Simulated work, independent of subscriptions. + await backend.append( + stream, event="progress", data={"number": number}, + producer="demo", idempotency_key=f"progress-{number}", + ) + jobs[job_id]["produced"] = number + await backend.finish( + stream, event="completed", data={"count": 20}, + producer="demo", idempotency_key="completed", + ) + jobs[job_id]["state"] = "completed" + except asyncio.CancelledError: + jobs[job_id]["state"] = "canceled" + raise + except Exception: + jobs[job_id]["state"] = "failed" + logging.exception("Demo event producer failed") + + +@asynccontextmanager +async def lifespan(app): + async with commands, relays, backend: + try: + yield + finally: + # Stop application-owned producers before closing the backend and clients. + pending = list(producers) + for task in pending: + task.cancel() + await asyncio.gather(*pending, return_exceptions=True) + + +app = FastAPI(lifespan=lifespan) +app.add_middleware(LocalRequestsOnly) +app.add_middleware(TrustedHostMiddleware, allowed_hosts=["127.0.0.1", "localhost"]) +add_redis_streams(app, backend=backend, manage_lifespan=False) + + +def get_job(job_id): + job = jobs.get(str(job_id)) + if job is None: + raise HTTPException(status_code=404, detail="Unknown demo job") + return job + + +@app.get("/", include_in_schema=False) +async def home(): + return FileResponse(Path(__file__).with_name("fastapi-sse-demo.html")) + + +@app.post("/jobs") +async def start_job(): + job_id = str(uuid4()) + # One fixed owner for this local demonstration; no request-supplied identity. + stream = f"local-demo-user:{job_id}" + await backend.create(stream, ttl=3600) + jobs[job_id] = {"stream": stream, "produced": 0, "state": "working"} + task = asyncio.create_task(produce(job_id, stream)) + producers.add(task) + task.add_done_callback(producers.discard) + return {"job_id": job_id} + + +@app.get("/jobs/{job_id}") +async def job_state(job_id: UUID): + job = get_job(job_id) + return {"job_id": str(job_id), "produced": job["produced"], "state": job["state"]} + + +@app.get( + "/jobs/{job_id}/events", + response_class=StreamingResponse, + responses={200: {"content": {"text/event-stream": {}}}}, +) +async def events(job_id: UUID, request: Request, streams: StreamBackendDep): + # This local demo has one owner. Authorize resource access in a deployed app. + job = get_job(job_id) + return await sse_response(request, backend=streams, stream=job["stream"]) + + +if __name__ == "__main__": + print("Open http://127.0.0.1:8000 in your browser.") + uvicorn.run(app, host="127.0.0.1", port=8000) From ced653bce66e11d72a07de4f1a0b78f32150b6c2 Mon Sep 17 00:00:00 2001 From: Julien Ruaux Date: Tue, 27 Oct 2026 16:43:17 +0100 Subject: [PATCH 2/9] docs(agent-mesh): add A2A and FastAPI SDK integration guides --- .../ai/context-engine/agent-mesh/AGENTS.md | 9 +- .../ai/context-engine/agent-mesh/_index.md | 2 +- .../ai/context-engine/agent-mesh/a2a-sdk.md | 180 ++++++++++++++++ .../agent-mesh/fastapi-sse-sdk.md | 201 ++++++++++++++++++ .../context-engine/agent-mesh/quickstart.md | 2 +- 5 files changed, 388 insertions(+), 6 deletions(-) create mode 100644 content/develop/ai/context-engine/agent-mesh/a2a-sdk.md create mode 100644 content/develop/ai/context-engine/agent-mesh/fastapi-sse-sdk.md diff --git a/content/develop/ai/context-engine/agent-mesh/AGENTS.md b/content/develop/ai/context-engine/agent-mesh/AGENTS.md index 519e556df0..941cd4b7a1 100644 --- a/content/develop/ai/context-engine/agent-mesh/AGENTS.md +++ b/content/develop/ai/context-engine/agent-mesh/AGENTS.md @@ -19,9 +19,10 @@ promise automatic execution restart or exactly-once external effects. - Keep the standalone Go mesh and its `agentmesh` HTTP client out of this SDK quickstart. Their deployment and Redis requirements are separate. -- Link detailed implementation and API guidance to redis-iris documentation. +- Keep user-facing integration guides here. Link exhaustive signatures and + implementation contracts to redis-iris documentation. Preserve the product preview banner and qualification limitations. -- Keep two public pages: an overview and one quickstart with both integrations. - Each example reaches one first result. Keep feature walkthroughs and detailed - configuration outside the quickstart. +- Keep an overview, one quickstart with both integrations, and one SDK guide + per integration. Each quickstart example reaches one first result. Keep + detailed configuration, authorization, replay and recovery in the SDK guides. diff --git a/content/develop/ai/context-engine/agent-mesh/_index.md b/content/develop/ai/context-engine/agent-mesh/_index.md index 7d13554fff..d8b610de59 100644 --- a/content/develop/ai/context-engine/agent-mesh/_index.md +++ b/content/develop/ai/context-engine/agent-mesh/_index.md @@ -62,4 +62,4 @@ For FastAPI, publish application events with `StreamBackend` and expose a stream Follow the [quickstart](/content/develop/ai/context-engine/agent-mesh/quickstart.md) to [resume A2A task updates](/content/develop/ai/context-engine/agent-mesh/quickstart.md#resume-a2a-task-updates) or [resume FastAPI events](/content/develop/ai/context-engine/agent-mesh/quickstart.md#resume-fastapi-events). -For detailed configuration and behavior, see the [A2A hosting guide](https://github.com/redis/redis-iris/blob/main/docs/user_guides/a2a-server.md) and [FastAPI integration guide](https://github.com/redis/redis-iris/blob/main/docs/user_guides/resumable-sse.md). +To integrate with your application, follow the [A2A SDK guide](/content/develop/ai/context-engine/agent-mesh/a2a-sdk.md) or [FastAPI SSE SDK guide](/content/develop/ai/context-engine/agent-mesh/fastapi-sse-sdk.md). These guides cover configuration, authorization, replay, and recovery. diff --git a/content/develop/ai/context-engine/agent-mesh/a2a-sdk.md b/content/develop/ai/context-engine/agent-mesh/a2a-sdk.md new file mode 100644 index 0000000000..6265bbce5d --- /dev/null +++ b/content/develop/ai/context-engine/agent-mesh/a2a-sdk.md @@ -0,0 +1,180 @@ +--- +alwaysopen: false +categories: +- docs +- develop +- ai +description: Add resumable task updates to an A2A Python application. +hideListLinks: true +linktitle: A2A SDK guide +title: Integrate Agent Mesh with A2A +weight: 20 +--- + +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](/content/develop/ai/context-engine/agent-mesh/quickstart.md#resume-a2a-task-updates). 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 or later with native producer idempotency (`XADD IDMP`), and the official A2A Python SDK 1.2.x. + +```bash +pip install 'redis-iris[a2a]' 'a2a-sdk[http-server]>=1.2.3,<1.3' +``` + +Use the [quickstart's database setup](/content/develop/ai/context-engine/agent-mesh/quickstart.md#choose-your-redis-database) 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. + +```python +import os +from contextlib import asynccontextmanager + +from a2a.server.request_handlers import DefaultRequestHandler +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 = DefaultRequestHandler( + agent_executor=store.heartbeat(agent_executor), + agent_card=agent_card, + task_store=store, + 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, 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 1,000 milliseconds. + +`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 + +`create_resumable_rest_routes()` 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/: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: + +```http +GET /tasks/:subscribe HTTP/1.1 +Accept: text/event-stream +A2A-Version: 1.0 +Authorization: Bearer +Last-Event-ID: +``` + +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](/content/develop/ai/context-engine/agent-mesh/quickstart.md#resume-a2a-task-updates). 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. + +## Handle replay and completion + +| Condition | Behavior | +| --- | --- | +| Retained cursor | Replay updates after the cursor, then follow new updates. | +| Trimmed history or compact Task replacement | Return the current Task snapshot, then follow events after its version. Intermediate updates covered by the snapshot are not reproduced. | +| Final cursor already consumed | Return HTTP 204; stop reconnecting. | +| No cursor for a terminal task | Reject the subscription; use the normal task retrieval API. | +| Malformed, repeated, or future cursor | Return HTTP 400. | +| Missing or expired snapshot | A new request receives TaskNotFound (HTTP 404). After streaming headers, send a TaskNotFound error frame. | + +The HTTP subscription ends when the Task becomes terminal or requires input or authentication. Handle the latter states through your application's A2A interaction before subscribing again. 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 + +The optional `store.heartbeat(executor, interval=5.0, alive_ttl=15.0)` wrapper refreshes liveness and retention while the agent executes. It checks stored terminal state and stops local execution when another replica cancels or finishes the Task. Set `alive_ttl` greater than `interval`. + +With an active store-backed subscriber, a lapsed heartbeat on a submitted or working Task marks it failed with a `producer_lost` status message. Polling task state alone does not trigger this recovery. Tasks waiting for input or authentication, and tasks without tracked heartbeats, are excluded. + +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. + +## Configure retention and limits + +Pass these options to `RedisTaskStore`: + +| Option | Default | Purpose | +| --- | --- | --- | +| `namespace` | `iris:a2a` | Isolate applications and deployments. | +| `active_ttl` | 604,800 seconds (7 days) | Active snapshot and event retention, refreshed by saves or heartbeats. | +| `terminal_ttl` | 86,400 seconds (1 day) | Terminal snapshot and event retention. | +| `metadata_ttl` | 86,400 seconds | Additional ownership and version retention after snapshot expiry. | +| `max_task_bytes` | 1,048,576 | Serialized snapshot size limit. | +| `max_event_bytes` | 1,048,576 | Serialized event size limit. | +| `max_events` | 10,000 | Approximate retained log length. | +| `max_attempts` | 5 | Maximum attempts under transaction contention. | + +The byte limits apply before hex encoding; stored payloads use about twice those serialized sizes. 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. + +| Public API | Use it to | +| --- | --- | +| `RedisTaskStore` | Store owner-scoped snapshots and versioned updates through the official SDK. | +| `store.event_stream()` | Build the matching event reader, with producer-loss recovery. | +| `RedisTaskEventStream` | Configure a reader explicitly with `store=store` and a separate client. | +| `store.heartbeat()` | Track execution liveness and notice remote cancellation. | +| `create_resumable_rest_routes()` | Expose cursor-enabled REST streaming. | +| `resumable_streams_extension()` | Advertise cursor support in the agent card. | + +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 the namespace and suits modest inventories. It is not a search index or a consistent snapshot during concurrent writes. + +See the [SDK API reference](https://github.com/redis/redis-iris/blob/main/docs/core/agent-mesh-a2a.md) for signatures and the [preview limits](/content/develop/ai/context-engine/agent-mesh/_index.md#requirements-and-limits) for deployment qualification. Redis durability remains application-owned; replay does not guarantee exactly-once external effects. diff --git a/content/develop/ai/context-engine/agent-mesh/fastapi-sse-sdk.md b/content/develop/ai/context-engine/agent-mesh/fastapi-sse-sdk.md new file mode 100644 index 0000000000..f15d005926 --- /dev/null +++ b/content/develop/ai/context-engine/agent-mesh/fastapi-sse-sdk.md @@ -0,0 +1,201 @@ +--- +alwaysopen: false +categories: +- docs +- develop +- ai +description: Add resumable application event streams to a FastAPI application. +hideListLinks: true +linktitle: FastAPI SSE SDK guide +title: Integrate Agent Mesh with FastAPI +weight: 30 +--- + +Use the Agent Mesh software development kit (SDK) to publish application events independently of client connections. FastAPI clients can reconnect with a saved cursor to replay retained server-sent events (SSE), then follow new ones. + +For a runnable browser example, start with the [FastAPI quickstart](/content/develop/ai/context-engine/agent-mesh/quickstart.md#resume-fastapi-events). This guide explains application integration, access checks, and recovery. + +## Install the integration + +You need Python 3.11 or later and Redis 8.6 or later with native producer idempotency (`XADD IDMP`). + +```bash +pip install 'redis-iris[fastapi-sse]' +``` + +Use the [quickstart's database setup](/content/develop/ai/context-engine/agent-mesh/quickstart.md#choose-your-redis-database) for a local or Redis Cloud connection. The A2A extra is independent and is not required. + +The `redis_iris.agent_mesh.fastapi_sse` module exposes the FastAPI Redis SDK streaming objects unchanged. Your application supplies configuration, Redis clients, authentication, and execution. + +## Register the backend + +Create separate Redis clients for commands and blocking event delivery. This example reads an application-owned `REDIS_URL` and supplies explicit `StreamSettings`. + +```python +import os +from contextlib import asynccontextmanager + +from fastapi import FastAPI +from redis.asyncio import Redis +from redis_iris.agent_mesh.fastapi_sse import ( + StreamBackend, + StreamSettings, + add_redis_streams, +) + +commands = Redis.from_url( + os.environ["REDIS_URL"], max_connections=16, socket_timeout=5 +) +relays = Redis.from_url( + os.environ["REDIS_URL"], max_connections=4, socket_timeout=5 +) +backend = StreamBackend( + commands, + blocking_client=relays, + settings=StreamSettings(prefix="myapp:sse", stream_bucket_count=4), +) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + async with commands, relays, backend: + yield + + +app = FastAPI(lifespan=lifespan) +add_redis_streams(app, backend=backend, manage_lifespan=False) +``` + +Register before startup so FastAPI installs the dependencies and error handlers. With `manage_lifespan=False`, your lifespan starts and stops the backend. It stops before the clients close. + +Alternatively, `add_redis_streams(app, backend=backend)` manages the backend around the existing application lifespan. In that mode, supplied clients must remain usable until backend shutdown. Registration never closes your clients. Give the backend exactly one lifecycle owner. + +Use independent pools connected to the same primary database with matching credentials and topology. The blocking pool needs at least one connection per configured bucket; Cluster pool limits apply per node. Set its socket timeout above `stream_block_ms`, which defaults to 1,000 milliseconds. Command and replay connections are additional. + +Configure credentials, Transport Layer Security (TLS), pools, and timeouts on your Redis clients. Explicit `StreamSettings` does not read `REDIS_*` variables or `.env` files, or configure other Iris clients. + +## Authorize the streaming endpoint + +Authenticate the caller and authorize the resource before accessing its stream. Neither `sse_response()` nor `StreamBackendDep` performs those checks. + +Add this endpoint to the application. `require_job_access` is your application dependency: it verifies the caller, loads the requested job in their tenant, checks access, and returns the job with its unique event-stream identity. + +```python +from fastapi import Depends, Request +from fastapi.responses import StreamingResponse +from redis_iris.agent_mesh.fastapi_sse import StreamBackendDep, sse_response + +from myapp.auth import require_job_access + + +@app.get( + "/jobs/{job_id}/events", + response_class=StreamingResponse, + responses={200: {"content": {"text/event-stream": {}}}}, +) +async def job_events( + request: Request, + streams: StreamBackendDep, + job=Depends(require_job_access), +): + return await sse_response(request, backend=streams, stream=job.event_stream_id) +``` + +Authorization runs on every reconnect. Stream IDs and cursors grant no access. Keep credentials out of URLs. + +Same-origin applications with cookie authentication can use native `EventSource`. Bearer-authenticated clients need a fetch-based SSE client or another client that supports authorization headers. You can pass `expires_at` to `sse_response()` using a timezone-aware deadline derived from verified credentials. It bounds the connection lifetime and reauthorization delay; it does not provide immediate revocation. + +## Publish progress and completion + +Create the event log when accepting work, then publish from your application workflow or a separately managed worker. The following excerpt assumes your application has allocated unique `stream_id` and `job_id` values: + +```python +await backend.create(stream_id, ttl=86400) +await backend.append( + stream_id, + event="job.progress", + data={"percent": 50}, + producer="job-worker", + idempotency_key=f"{job_id}:progress:50", +) +await backend.finish( + stream_id, + event="job.completed", + data={"job_id": job_id}, + producer="job-worker", + idempotency_key=f"{job_id}:completed", +) +``` + +`finish()` seals the log with a final event. Later appends fail. A matching finish retry returns the retained final cursor. Use a distinct retry identity for completion. + +Keep producer and event identities stable across retries, with the same payload. Deduplication defaults to 24 hours and 10,000 IDs per producer, bounded by stream lifetime. It does not compare payloads or deduplicate external effects. A retry of a trimmed event can return its old cursor without restoring the payload. + +Disconnecting a subscriber does not cancel the producer. The SDK does not schedule or restart jobs. Publisher-only workers can use `StreamBackend(client, settings=settings)` and call `await backend.initialize()` without starting a relay runtime. Workers and serving replicas must use the same prefix, bucket layout, serialization, and stream settings. + +## Resume delivery + +Stored events have Redis entry IDs. Save each cursor after applying its event. A reconnect with `Last-Event-ID` receives events strictly after that cursor. Native `EventSource` resends its last ID on automatic reconnects; for a newly created connection, supply your saved cursor, for example with the `after` query parameter. + +Cursor selection is explicit `sse_response(after=...)`, then `Last-Event-ID`, then query `after`, then `0-0`. Every supplied value is validated, even when another takes precedence. Empty, duplicate, malformed, special Redis, or future cursors fail. + +A reconnect after the final event returns HTTP 204. Stop reconnecting once completion is established. An ended connection alone does not prove the job completed. + +## Recover when history is unavailable + +Redis retains a delivery window, not your application's permanent state. If a client misses more history than the log retains, replay cannot reconstruct the missing changes. + +Provide an authenticated application endpoint that returns current job state together with a matching event cursor. On recovery, the client replaces its local state with that result, saves its cursor, and resumes after it. Ensure the state and cursor describe the same point in the job's progress; otherwise updates can be skipped or applied twice. + +If the old log has expired, your application decides whether to return a completed result or establish a new delivery window with a new stream identity. Do not silently jump to the retained tail of the old stream. + +Before streaming headers are sent, registered handlers return errors with a stable `detail.code`: + +| HTTP status | Code or condition | Client action | +| --- | --- | --- | +| 404 | `stream_not_found` | Retrieve application state; the stream is unknown. | +| 410 | `stream_expired` | Retrieve application state; the delivery window expired. | +| 409 | `history_unavailable` | Reload state with a matching cursor. | +| 409 | `cursor_ahead` | Reconcile state and cursor; do not retry the same cursor blindly. | +| 409 | `stream_conflict` | Resolve the lifecycle conflict. | +| 422 | `invalid_stream_request` | Correct the request. | +| 503 | `stream_capacity_exceeded` or `stream_unavailable` | Retry with backoff, preserving the last applied cursor. | +| 204 | Final event already consumed | Stop reconnecting. | + +Your authentication dependencies retain their own status and challenge headers. + +After headers, recoverable failures send a `redis.stream.control` event when possible, then close. Follow its `action`: `reload`, `reconnect`, or `reauthenticate`. Control frames and keepalives have no event ID and must not advance the cursor. A broken socket might receive no control frame. Native `EventSource` cannot inspect initial error bodies; use your application state endpoint or a fetch-based client for recovery. + +## Configure retention and resources + +`create(stream_id, ttl=...)` sets an absolute expiration. Reads, writes, and creation retries do not extend it. After payload expiry, content-free metadata recognizes expiry for a further window; after metadata expiry, the stream is unknown. Allocate unique stream identities across tenants and delivery lifetimes. + +Pass settings as lowercase fields on `StreamSettings`: + +| Setting | Default | Purpose | +| --- | --- | --- | +| `prefix` | Set explicitly in the example | Isolate a storage namespace; do not include braces. | +| `stream_bucket_count` | 16 | Fixed placement layout for the namespace. | +| `stream_max_event_bytes` | 65,536 | Maximum encoded storage bytes per event. | +| `stream_max_events` | 10,000 | Retained application events. | +| `stream_idmp_duration` / `stream_idmp_maxsize` | 86,400 seconds / 10,000 | Deduplication window and IDs per producer. | +| `stream_expiry_recognition` | 86,400 seconds | Expired-stream recognition window. | +| `stream_max_duration` / `stream_keepalive` | 300 / 15 seconds | Connection lifetime and keepalive interval. | +| `stream_queue_events` / `stream_queue_bytes` | 256 / 1,048,576 | Per-subscriber queue bounds. | +| `stream_max_subscribers` / `stream_max_active_streams` | 1,000 / 256 | Per-process admission bounds. | +| `stream_block_ms` | 1,000 | Blocking read interval in milliseconds. | + +Do not change the persisted bucket layout in place. Use a new namespace for a different layout. Delivery uses one blocking reader per active bucket per process, with bounded subscriber queues and replay admission. Slow consumers can be disconnected and must recover with their saved cursor. + +| Public API | Use it to | +| --- | --- | +| `StreamBackend` | Create, append, finish, replay, and follow event logs. | +| `StreamSettings` | Configure explicit storage and delivery limits. | +| `StreamEvent` | Read an event's `id`, `event`, `data`, and `final` metadata. | +| `StreamBackendDep`, `get_stream_backend` | Access the registered backend from FastAPI. | +| `add_redis_streams` | Register dependencies, error handlers, and optional lifespan management. | +| `sse_response` | Prepare replay and live delivery after application authorization. | + +Catch the exported `StreamError` subclasses for direct backend calls, including `StreamNotFound`, `StreamExpired`, `StreamConflict`, `StreamHistoryUnavailable`, `StreamCursorAhead`, `StreamCapacityExceeded`, `StreamSlowConsumer`, and `InvalidStreamRequest`. The registered HTTP handlers provide the client responses described in [recovery](/content/develop/ai/context-engine/agent-mesh/fastapi-sse-sdk.md#recover-when-history-is-unavailable). + +See the [SDK reference](https://github.com/redis/redis-iris/blob/main/docs/core/agent-mesh-sse.md) for the complete export inventory and upstream API documentation, and the [preview limits](/content/develop/ai/context-engine/agent-mesh/_index.md#requirements-and-limits) for deployment qualification. diff --git a/content/develop/ai/context-engine/agent-mesh/quickstart.md b/content/develop/ai/context-engine/agent-mesh/quickstart.md index 178fe7cc0b..1ca7942e36 100644 --- a/content/develop/ai/context-engine/agent-mesh/quickstart.md +++ b/content/develop/ai/context-engine/agent-mesh/quickstart.md @@ -227,4 +227,4 @@ Stopping the app leaves a Redis Cloud database running. If you created a disposa ## Further reading -For configuration and behavior beyond these first examples, see the [A2A hosting guide](https://github.com/redis/redis-iris/blob/main/docs/user_guides/a2a-server.md), [FastAPI integration guide](https://github.com/redis/redis-iris/blob/main/docs/user_guides/resumable-sse.md), and [Agent Mesh limitations](/content/develop/ai/context-engine/agent-mesh/_index.md#requirements-and-limits). +To add these integrations to your application, follow the [A2A SDK guide](/content/develop/ai/context-engine/agent-mesh/a2a-sdk.md) or [FastAPI SSE SDK guide](/content/develop/ai/context-engine/agent-mesh/fastapi-sse-sdk.md). See the [Agent Mesh limitations](/content/develop/ai/context-engine/agent-mesh/_index.md#requirements-and-limits) for preview qualification. From 5a16d4437659736a5f996d658f603ea59dcaf309 Mon Sep 17 00:00:00 2001 From: Julien Ruaux Date: Wed, 28 Oct 2026 09:31:08 +0100 Subject: [PATCH 3/9] docs(agent-mesh): demonstrate agent progress in FastAPI quickstart --- .../context-engine/agent-mesh/quickstart.md | 18 +++---- static/code/agent-mesh/fastapi-sse-demo.html | 50 ++++++++++++------- static/code/agent-mesh/fastapi-sse-demo.py | 39 +++++++++++++-- 3 files changed, 77 insertions(+), 30 deletions(-) diff --git a/content/develop/ai/context-engine/agent-mesh/quickstart.md b/content/develop/ai/context-engine/agent-mesh/quickstart.md index 1ca7942e36..0a6b791bde 100644 --- a/content/develop/ai/context-engine/agent-mesh/quickstart.md +++ b/content/develop/ai/context-engine/agent-mesh/quickstart.md @@ -139,7 +139,7 @@ Stop the application with Control-C before trying the FastAPI example on the sam ## Resume FastAPI events -An application job publishes twenty progress events, one per second. Disconnect its subscription while the job continues, then reconnect to receive the events you missed. These are application-defined events, independent of A2A tasks. +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 @@ -172,16 +172,16 @@ This loopback-only example uses one fixed demo owner and needs no token or sign- ### Disconnect and catch up -1. Select **Start job**. Numbered progress events appear in the page. +1. Select **Start agent**. Progress messages show the agent preparing its itinerary. 1. After a few events arrive, select **Disconnect**. The displayed event count stops. -1. Wait a few seconds, then select **Reconnect**. The missed events arrive, followed by new events as the job continues. -1. Wait for **Complete — all 20 events received**. +1. Wait a few seconds, then select **Reconnect**. The missed events arrive, followed by new events as the agent continues. +1. Wait for **Complete — all 20 events received** and the agent’s sample itinerary. -The page saves its event cursor and reconnects to the same job's event stream. Expand **What is being resumed?** to see the job ID and saved cursor. The job continues independently of the subscription; reconnecting does not start it again. +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 owns its Redis clients, startup, and job execution. Its streaming backend uses separate command and relay clients: +The downloaded application owns its Redis clients, startup, and simulated agent execution. Its streaming backend uses separate command and relay clients: ```python from redis_iris.agent_mesh.fastapi_sse import StreamBackend, StreamSettings @@ -192,16 +192,16 @@ backend = StreamBackend( ) ``` -Create a log when starting a job, append its progress events, and finish it when work completes: +Create a log when starting the agent, append its progress messages, and finish with its answer: ```python await backend.create(stream, ttl=3600) await backend.append( - stream, event="progress", data={"number": number}, + stream, event="progress", data={"number": number, "message": message}, producer="demo", idempotency_key=f"progress-{number}", ) await backend.finish( - stream, event="completed", data={"count": 20}, + stream, event="completed", data={"count": len(AGENT_STEPS), "answer": AGENT_ANSWER}, producer="demo", idempotency_key="completed", ) ``` diff --git a/static/code/agent-mesh/fastapi-sse-demo.html b/static/code/agent-mesh/fastapi-sse-demo.html index 762b3b386a..8f0dabccda 100644 --- a/static/code/agent-mesh/fastapi-sse-demo.html +++ b/static/code/agent-mesh/fastapi-sse-demo.html @@ -28,13 +28,15 @@
      AGENT MESH · LOCAL FASTAPI SSE DEMO
      -

      Reconnect without missing updates

      -

      An application job publishes 20 progress events, one per second. Disconnect your subscription, wait a few seconds, and reconnect. The job keeps working while you are away.

      +

      Resume your agent’s progress

      +

      A simulated travel agent works on your request and streams its progress. Disconnect, wait a few seconds, and reconnect to catch up and receive the answer. The agent keeps working while you are away.

      +

      Your request: “Plan a day in Rome with sightseeing, a museum, and time for breaks.”

      +

      The agent uses scripted steps and a sample answer. No model account or live travel lookup is required.

      Ready to start
      -

      Start a job, then disconnect after a few updates arrive.

      +

      Start the agent, then disconnect after a few updates arrive.

      - +
      @@ -42,17 +44,21 @@

      Reconnect without missing updates

        +
        What is being resumed? -

        Job: —
        Last applied cursor: —

        +

        Agent run: —
        Last applied cursor: —

        The page saves each event cursor after applying its update. Reconnect uses ?after= to request the retained events after that cursor. The browser's native EventSource receives the stream.

        -

        Local demonstration only. There is no sign-in: all requests use one fixed demo owner. Closing a subscription does not cancel the application job. Replay depends on retained Redis history.

        +

        Local demonstration only. There is no sign-in: all requests use one fixed demo owner. Your application runs the simulated agent; Agent Mesh retains and delivers its events. Closing a subscription does not cancel the agent. Replay depends on retained Redis history.