diff --git a/content/develop/ai/context-engine/_index.md b/content/develop/ai/context-engine/_index.md index ef97b98428..562ebd0028 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 agent-to-agent 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..f7292eddca --- /dev/null +++ b/content/develop/ai/context-engine/agent-mesh/AGENTS.md @@ -0,0 +1,31 @@ +# 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. Both extras can be + installed together; their shared redis-py range is `>=8.0.1,<9`. Keep Redis + server requirements (A2A 8.6+, FastAPI 8.6.2+) distinct from Python client versions. +- 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 supply authentication and agent execution. A2A accepts application-owned + Redis clients; FastAPI uses SDK-managed pools and producer 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. +- 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 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 new file mode 100644 index 0000000000..ab7c2a57f0 --- /dev/null +++ b/content/develop/ai/context-engine/agent-mesh/_index.md @@ -0,0 +1,66 @@ +--- +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 +--- + +## What is Redis Agent Mesh? +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 supplies agent execution and authentication. A2A accepts application-owned Redis clients; the FastAPI integration uses SDK-managed pools and producer 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, or install both in one Python environment with `pip install 'redis-iris[a2a,fastapi-sse]'`. Using both selects redis-py `>=8.0.1,<9`. This is the Python client version; A2A requires Redis server 8.6+; FastAPI requires 8.6.2+. + +## 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, configure `FastAPIRedis(app).lifespan().streams()`, publish scoped run events through a `StreamWriter`, and serve them with `backend.sse()`. Event production is independent of client connections. + +## Requirements and limits + +- Both integrations require Python 3.11 or later and native producer idempotency (`XADD IDMP`). A2A requires Redis 8.6+ and FastAPI requires 8.6.2+. 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). + +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..d80495c681 --- /dev/null +++ b/content/develop/ai/context-engine/agent-mesh/a2a-sdk.md @@ -0,0 +1,184 @@ +--- +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' +``` + +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](/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. | + +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 + +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..30c034b36f --- /dev/null +++ b/content/develop/ai/context-engine/agent-mesh/fastapi-sse-sdk.md @@ -0,0 +1,214 @@ +--- +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 run a producer independently +of its client connection. FastAPI clients reconnect with a saved cursor to replay +retained server-sent events (SSE), then follow new events from the same run. + +Start with the [FastAPI browser demo](/content/develop/ai/context-engine/agent-mesh/quickstart.md#resume-fastapi-events). +This guide covers application setup, authorization, execution and recovery. + +## Install the integration + +Use Python 3.11 or later and Redis 8.6.2 or later with native producer idempotency +(`XADD IDMP`). Redis 8.6.6 or later is recommended by the upstream streaming guide. + +```bash +pip install 'redis-iris[fastapi-sse]' +``` + +The independent A2A integration is optional. Install both with +`redis-iris[a2a,fastapi-sse]`; their shared redis-py range is `>=8.0.1,<9`. +See the [database setup](/content/develop/ai/context-engine/agent-mesh/quickstart.md#choose-your-redis-database) +for local Redis or Redis Cloud connections. Deployment qualification is separate +from configuring an endpoint. + +## Configure the application + +The facade re-exports the upstream FastAPI Redis SDK objects unchanged. Configure +its standard lifespan and streaming feature before application startup: + +```python +from fastapi import FastAPI +from redis_iris.agent_mesh.fastapi_sse import FastAPIRedis + +app = FastAPI() +FastAPIRedis(app).lifespan().streams() +``` + +Set `REDIS_URL` for the connection, including any Transport Layer Security (TLS) +parameters. Set `REDIS_PREFIX` for the namespace. The SDK creates and closes its +Redis pools and stops its in-process producers when the application shuts down. +Use its `REDIS_*` configuration and `.env` sources; settings are cached globally. +The integration does not accept injected clients, per-app settings or a separate +blocking-client argument. + +Keep `REDIS_SOCKET_TIMEOUT` unset, or longer than +`REDIS_STREAM_HEARTBEAT_INTERVAL` (15 seconds by default), so blocking reads can +complete. Each live reader holds one connection from the shared pool. If you set +`REDIS_MAX_CONNECTIONS`, reserve capacity for readers plus writes and other +commands. There are no relay buckets or subscriber queues. + +## Authorize scoped runs + +A run is identified by a `scope` and `stream_id`. The scope also defines the +exclusive active-run boundary: by default, only one run is active per scope. +Choose a resource-sized scope, such as a tenant's chat or job, rather than using +one scope for every job in a tenant. + +Authenticate and authorize before passing the scope and run ID to the backend. +Neither `StreamBackendDep` nor `backend.sse()` performs application access checks. +The application dependency in this example verifies access to the job and returns +its authorized tenant and run identifiers: + +```python +from fastapi import Depends, Request +from fastapi.responses import StreamingResponse +from redis_iris.agent_mesh.fastapi_sse import StreamBackendDep + +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), +): + scope = f"tenant:{job.tenant_id}:job:{job.id}" + return await streams.sse(job.run_id, request=request, scope=scope) +``` + +Authorize every reconnect. IDs and cursors grant no access. Scopes accept +letters, digits, underscores, periods, colons and hyphens; run IDs accept letters, +digits, underscores and hyphens. Keep credentials out of URLs. + +Same-origin cookie authentication works with native `EventSource`. Bearer tokens +need a fetch-based SSE client or another client that accepts authorization +headers. Pass a timezone-aware verified credential deadline as `expires_at` to +`backend.sse()` to bound the open connection. This does not provide immediate +revocation. + +## Start a producer and publish events + +Supply an async producer that receives a `StreamWriter`: + +```python +from redis_iris.agent_mesh.fastapi_sse import StreamWriter + + +async def run_job(writer: StreamWriter): + result = await perform_job() + await writer.append("job.progress", {"percent": 100}) + await writer.finish("job.completed", {"result": result}) + + +# In your authorized start endpoint, with a StreamBackendDep named streams: +# scope and run_id come from the application's job record. +writer = await streams.start(run_job, scope=scope, stream_id=run_id) +``` + +`start()` opens the run and starts the producer only when it was newly created. +A repeated start with the same scope and retained ID returns a retry handle +without spawning a second producer. `writer.append(type, data)` and +`writer.finish(type, data)` return Redis entry IDs. Event types are application +names; the `stream.` prefix is reserved for SDK events. + +For additional work between opening and spawning, use `await streams.open(...)` +and `streams.spawn(writer, producer)`. To read without HTTP, use +`async with await streams.follow(run_id, scope=scope, after=cursor)`. +Events expose `id`, `type`, `data` and `final` fields. + +The SDK owns the producer task and generates its write-retry identities. +Applications cannot supply producer/event idempotency keys. Disconnecting a +reader leaves the task running. `cancel()` requests a stop; `active()` identifies +the scope's active run; `attach()` takes over an unfinished run. These operations +do not undo external side effects or restart computation after a process exit. +Publisher-only external workers are not supported in this version. + +## Resume and recover + +Save each event cursor after applying its update. Reconnect to the same scope +and run with `Last-Event-ID` or `?after=`. Selection order is explicit +`backend.sse(after=...)`, then the header, then the query, then the start of the +stream. Every supplied cursor is validated; repeated or malformed values fail. +A reconnect at the final cursor returns HTTP 204. + +Retention bounds replay. Provide an authorized application endpoint returning +current state together with a matching cursor when clients need recovery. +Reload state when reconnecting returns HTTP 410 for expired or trimmed history. +A `stream.gap` event during delivery reports missing entries and the reader +continues; a client requiring complete history should recover application state. + +Before headers, errors have the shape `{"detail": "...", "code": "..."}`. +Read the top-level `code` field: + +| HTTP status | Code or condition | Client action | +| --- | --- | --- | +| 401 | `credential_expired` | Refresh verified credentials. | +| 404 | `stream_not_found` | Retrieve application state. | +| 410 | `stream_expired` | Reload state; history expired or was trimmed. | +| 409 | `stream_busy` | Reuse the active run or resolve the scope conflict. | +| 409 | `cursor_ahead` | Reconcile application state and cursor. | +| 409 | `stream_fenced`, `stream_finished` | Stop using the old writer. | +| 422 | `invalid_cursor`, `invalid_stream_name` | Correct the request. | +| 503 | `stream_unavailable`, `stream_unsupported`, or reader capacity | Retry transient failures with backoff; check server requirements for unsupported features. | +| 204 | Final event consumed | Stop reconnecting. | + +Authentication dependencies retain their own status and challenge headers. +After headers, transient failures close the connection and clients reconnect. +Handle SDK events `stream.end` (including `error`, `cancelled` and +`producer_lost` outcomes) and `stream.error` (credential expiry or read-only +producer-loss reporting). These replace control-action frames. Native +`EventSource` cannot inspect refused-connection bodies; use the application +state endpoint or a fetch-based client to explain failures. + +## Configure retention + +Retention slides on writes. Active appends reset `stream_active_ttl`; completion +sets `stream_done_ttl`; metadata outlives payload by `stream_gone_ttl`. +Reads do not extend retention. `writer.extend(ttl)` extends a waiting run's +retention. `spawn()` runs `writer.keepalive()` around the producer, or an +application-managed producer can enter that context itself. + +Keep idle waits below active retention or arrange explicit extensions. Liveness +and retained history are separate: a heartbeat alone does not guarantee payload +retention in the current version. Allow for the heartbeat's own active-slot TTL +when extending a waiting run. A producer lost on a process exit is not restarted. + +| Setting (prefix with `REDIS_`) | Default | Purpose | +| --- | --- | --- | +| `STREAM_ACTIVE_TTL` | 600 seconds | Retention refreshed by active writes. | +| `STREAM_DONE_TTL` | 120 seconds | Replay window after completion. | +| `STREAM_GONE_TTL` | 3,600 seconds | Expired-history recognition window. | +| `STREAM_MAXLEN` | 10,000 | Approximate retained entries. | +| `STREAM_MAX_EVENT_BYTES` | 65,536 | Largest encoded event. | +| `STREAM_HEARTBEAT_INTERVAL` | 15 seconds | Producer heartbeat, SSE ping and blocking-read bound. | +| `STREAM_ALIVE_TTL` | 45 seconds | Producer-liveness field lifetime. | +| `STREAM_MAX_READERS` | 256 | Live-reader admission per process. | +| `STREAM_REPLAY_PAGE` | 100 | Events per Redis read. | +| `STREAM_MAX_CONNECTION_DURATION` | 300 seconds | Connection cap before reconnect. | +| `STREAM_READ_RETRY_WINDOW` | 30 seconds | Retry window for transient read failures. | +| `STREAM_SEND_TIMEOUT` | 10 seconds | Bound for stalled sends. | + +Set `REDIS_STREAM_DONE_TTL=3600` to retain a completed demo run for one hour. +These settings differ from the application state retained in its own database. + +See the [SDK reference](https://github.com/redis/redis-iris/blob/main/docs/core/agent-mesh-sse.md) +for facade exports and upstream signatures, 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 new file mode 100644 index 0000000000..c03e530956 --- /dev/null +++ b/content/develop/ai/context-engine/agent-mesh/quickstart.md @@ -0,0 +1,231 @@ +--- +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.2 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.2 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 + +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: + +```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 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 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 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: + +```python +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: + +```python +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: + +```python +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 +one hour. Active retention and finished retention are separate. 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](/content/operate/rc/databases/delete-database.md). + +## Further reading + +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. diff --git a/content/operate/iris/_index.md b/content/operate/iris/_index.md index 11cb775d07..16ebe6b460 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 agent-to-agent 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..bd31f0f070 --- /dev/null +++ b/static/code/agent-mesh/fastapi-sse-demo.html @@ -0,0 +1,211 @@ + + + + + + Agent Mesh: reconnect without missing updates + + + +
    +
    AGENT MESH · LOCAL FASTAPI SSE DEMO
    +

    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 the agent, then disconnect after a few updates arrive.

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

      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. 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.

      +
      + + + 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..cfcb5d417b --- /dev/null +++ b/static/code/agent-mesh/fastapi-sse-demo.py @@ -0,0 +1,132 @@ +"""Loopback-only simulated agent demo with FastAPI events. Run: python fastapi-sse-demo.py.""" + +import asyncio +import os +import logging +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_iris.agent_mesh.fastapi_sse import FastAPIRedis, StreamBackendDep +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") +# The upstream SDK owns its pools and in-process producer tasks. +os.environ.setdefault("REDIS_PREFIX", "quickstart:sse-browser") +os.environ.setdefault("REDIS_STREAM_DONE_TTL", "3600") +jobs = {} + + +# Scripted agent activity keeps this example runnable without a model account. +AGENT_STEPS = ( + "Reading your request", + "Identifying the trip length and interests", + "Listing candidate sights", + "Looking up the historic center in the sample travel notes", + "Comparing morning walking routes", + "Selecting a starting point", + "Checking the sample museum information", + "Allowing time for a museum visit", + "Looking up lunch options in the sample notes", + "Choosing a lunch stop near the route", + "Planning the afternoon walk", + "Adding a rest break", + "Comparing evening viewpoints", + "Selecting a sunset stop", + "Estimating time between stops", + "Checking the itinerary for backtracking", + "Balancing sightseeing and breaks", + "Drafting the itinerary", + "Adding a reminder to check opening times", + "Preparing your answer", +) +AGENT_ANSWER = ( + "Spend the morning walking from the Pantheon to Piazza Navona. " + "Have lunch nearby, then visit the Capitoline Museums in the afternoon. " + "Take a coffee break before walking to the Orange Garden for an evening view. " + "Check opening times and ticket availability before your visit." +) + + +async def produce(job_id, writer): + try: + for number, message in enumerate(AGENT_STEPS, start=1): + await asyncio.sleep(1) # Simulated work, independent of subscriptions. + await writer.append("progress", {"number": number, "message": message}) + jobs[job_id]["produced"] = number + await writer.finish("completed", {"count": len(AGENT_STEPS), "answer": AGENT_ANSWER}) + 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") + raise + + +app = FastAPI() +FastAPIRedis(app).lifespan().streams() +app.add_middleware(LocalRequestsOnly) +app.add_middleware(TrustedHostMiddleware, allowed_hosts=["127.0.0.1", "localhost"]) + + +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(streams: StreamBackendDep): + job_id = str(uuid4()) + # One fixed owner for this local demonstration; no request-supplied identity. + scope = f"job:{job_id}" + jobs[job_id] = {"scope": scope, "produced": 0, "state": "working"} + await streams.start(lambda writer: produce(job_id, writer), scope=scope, stream_id=job_id) + 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 streams.sse(str(job_id), request=request, scope=job["scope"]) + + +if __name__ == "__main__": + print("Open http://127.0.0.1:8000 in your browser.") + uvicorn.run(app, host="127.0.0.1", port=8000)