Metadata-Version: 2.4
Name: fastapi-resumable-stream
Version: 0.2.0
Summary: Resumable SSE streams for FastAPI — survive refresh, reconnects, and rolling deploys
Project-URL: Homepage, https://github.com/ofershap/fastapi-resumable-stream
Project-URL: Repository, https://github.com/ofershap/fastapi-resumable-stream
Project-URL: Issues, https://github.com/ofershap/fastapi-resumable-stream/issues
Author: Ofer Shapira
License-Expression: MIT
License-File: LICENSE
Keywords: fastapi,llm,redis,resumable,sse,streaming
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Internet :: WWW/HTTP
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: redis>=5.0.0
Provides-Extra: dev
Requires-Dist: fakeredis[lua]>=2.23.0; extra == 'dev'
Requires-Dist: fastapi>=0.110.0; extra == 'dev'
Requires-Dist: httpx>=0.27.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Requires-Dist: ruff>=0.8.0; extra == 'dev'
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.110.0; extra == 'fastapi'
Description-Content-Type: text/markdown

# fastapi-resumable-stream

[![PyPI version](https://img.shields.io/pypi/v/fastapi-resumable-stream.svg)](https://pypi.org/project/fastapi-resumable-stream/)
[![CI](https://github.com/ofershap/fastapi-resumable-stream/actions/workflows/ci.yml/badge.svg)](https://github.com/ofershap/fastapi-resumable-stream/actions/workflows/ci.yml)
[![Python](https://img.shields.io/badge/python-3.10%2B-blue.svg)](https://www.python.org/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

Your user refreshes the page mid-stream and the LLM response is gone. JS has `resumable-stream` for this. Python doesn't have a default — until now.

```python
stream = ResumableStream(redis)

# Start a stream: producer keeps running even if the client disconnects
async for chunk in await stream.start("stream-1", producer=my_llm_generator):
    yield chunk

# Reconnect after refresh — replays what was missed, then keeps streaming
resumed = await stream.resume("stream-1", after=chunks_already_seen)
```

> Redis-backed. No new infra, no vendor lock-in, no SaaS bill — a `redis.asyncio.Redis` client is the only dependency. Ships FastAPI helpers for the common case: start / resume / stop tied to a conversation ID.

## Why

Vercel's [`resumable-stream`](https://github.com/vercel/resumable-stream) is the default for Next.js: disconnect, refresh, reconnect — the stream picks up where it left off because the producer keeps running server-side in Redis, decoupled from any single HTTP connection. Python backends running LangChain, CrewAI, or a raw OpenAI/Anthropic loop on FastAPI don't have an equivalent default. This is that default, purpose-built for FastAPI's `StreamingResponse`.

## Install

```bash
pip install fastapi-resumable-stream[fastapi]
```

Without FastAPI (framework-agnostic core only):

```bash
pip install fastapi-resumable-stream
```

## Usage

A minimal FastAPI chat endpoint that survives a page refresh:

```python
from fastapi import FastAPI
from redis.asyncio import Redis

from fastapi_resumable_stream import (
    ActiveStreamRegistry,
    ResumableStream,
    StopController,
    resume_conversation_stream,
    start_conversation_stream,
    stop_conversation_stream,
    streaming_response,
)

app = FastAPI()
redis = Redis.from_url("redis://localhost:6379")
stream = ResumableStream(redis)
registry = ActiveStreamRegistry(redis)
stop = StopController(redis)


async def call_llm(prompt: str):
    async for token in my_llm_client.stream(prompt):
        yield token


@app.post("/chat/{conversation_id}/stream")
async def start_stream(conversation_id: str, prompt: str):
    stream_id = f"{conversation_id}:{prompt_hash(prompt)}"
    return await start_conversation_stream(
        stream, registry, stop, conversation_id, stream_id,
        producer=lambda: call_llm(prompt),
    )


@app.get("/chat/{conversation_id}/stream")
async def resume_stream(conversation_id: str, after: int = 0):
    resumed = await resume_conversation_stream(stream, registry, conversation_id, after=after)
    if resumed is None:
        return Response(status_code=204)  # nothing to resume
    return streaming_response(resumed)


@app.post("/chat/{conversation_id}/stop")
async def stop_stream(conversation_id: str):
    stopped = await stop_conversation_stream(registry, stop, conversation_id)
    return {"stopped": stopped}
```

On the client: keep a running count of chunks received, and on reconnect call `GET /chat/{id}/stream?after=<count>` instead of restarting the request.

## How it works

- **Producer runs independently of the HTTP connection.** `ResumableStream.start()` launches the producer as a background `asyncio.Task`. If the client disconnects, the task keeps writing chunks to Redis — it isn't tied to the request lifecycle.
- **Chunks are buffered in a Redis list**, one entry per chunk, with a pub/sub channel used to wake up waiting consumers instead of pure polling.
- **Resuming replays from an index**, not a byte offset — pass how many chunks you've already seen (`after`) and get everything since, then keep streaming until done.
- **Disconnect isn't cancel.** A dropped HTTP connection stops a consumer, not the producer. Call `stop_conversation_stream()` to actually cancel generation (e.g. a user-pressed stop button).
- **`ActiveStreamRegistry`** maps a `conversation_id` to whatever `stream_id` is currently running, so your resume/stop endpoints don't need the client to remember stream IDs across page loads.

## Framework recipes

The core only cares about `AsyncIterator[str]` — it doesn't know or care what's inside a chunk. That means wiring up a specific agent framework or frontend wire protocol is a thin adapter, not a rewrite.

### LangChain

```python
from fastapi_resumable_stream.adapters import from_langchain_events

producer = lambda: from_langchain_events(chain.astream_events(inputs, version="v2"))
```

Filters `astream_events()` down to `on_chat_model_stream` token deltas.

### CrewAI

```python
from fastapi_resumable_stream.adapters import from_crewai_stream

async def producer():
    streaming = await crew.akickoff(inputs=inputs)
    async for chunk in from_crewai_stream(streaming):
        yield chunk
```

> As of CrewAI's current release, `CrewStreamingOutput` has no `aclose()`/`cancel()` ([crewAIInc/crewAI#5312](https://github.com/crewAIInc/crewAI/issues/5312)). `stop_conversation_stream()` will correctly stop relaying chunks to your client, but the underlying crew run keeps consuming tokens/compute until CrewAI ships that fix.

### AG-UI protocol

```python
from fastapi_resumable_stream.adapters import to_ag_ui_text_message

producer = lambda: to_ag_ui_text_message(my_token_producer())
```

Wraps a plain token stream in the `TEXT_MESSAGE_START` / `TEXT_MESSAGE_CONTENT` / `TEXT_MESSAGE_END` triad ([AG-UI events spec](https://docs.ag-ui.com/concepts/events)). Zero-dependency — hand-encodes the JSON directly rather than pinning to the `ag-ui-protocol` package.

### Vercel AI SDK (`useChat`)

```python
from fastapi_resumable_stream.adapters import VERCEL_STREAM_HEADERS, to_vercel_ui_message_stream

@app.post("/chat/{conversation_id}/stream")
async def start_stream(conversation_id: str, prompt: str):
    stream_id = f"{conversation_id}:{prompt_hash(prompt)}"
    return await start_conversation_stream(
        stream, registry, stop, conversation_id, stream_id,
        producer=lambda: to_vercel_ui_message_stream(my_token_producer()),
        extra_headers=VERCEL_STREAM_HEADERS,
    )
```

Targets the current (AI SDK v7-era) UI Message Stream Protocol: `start` → `text-start` → `text-delta`* → `text-end` → `finish` → `[DONE]`. This wire format has changed across AI SDK v4/v5/v6/v7 — if `useChat` stops parsing the stream after a Vercel upgrade, check the [stream protocol docs](https://ai-sdk.dev/docs/ai-sdk-ui/stream-protocol) for the current event shape.

## API

### `ResumableStream(redis, *, prefix="fastapi-resumable-stream", ttl=3600, poll_interval=0.05)`

- `await start(stream_id, producer) -> AsyncIterator[str]` — starts (or attaches to) a stream, returning an iterator of chunks from the beginning. Marks the stream active and launches the producer *before* returning, so it isn't at the mercy of whether the result ever gets iterated.
- `resume(stream_id, *, after=0) -> AsyncIterator[str] | None` — `None` if the stream doesn't exist; otherwise yields chunks from index `after` onward, live if still active.
- `status(stream_id) -> Literal["missing", "active", "done"]`
- `close()` — cancels any producer tasks owned by this instance (call on shutdown).

### `ActiveStreamRegistry(redis, *, prefix="fastapi-resumable-stream")`

- `set_active(conversation_id, stream_id, *, ttl=3600)`
- `get_active(conversation_id) -> str | None`
- `clear(conversation_id)`

### `StopController(redis, *, prefix="fastapi-resumable-stream")`

- `request_stop(stream_id, *, ttl=3600)`
- `is_stopped(stream_id) -> bool`
- `clear(stream_id)`

### FastAPI helpers

- `streaming_response(iterator, *, extra_headers=None) -> StreamingResponse` — wraps an async iterator with the right SSE headers.
- `start_conversation_stream(stream, registry, stop, conversation_id, stream_id, producer, *, extra_headers=None) -> StreamingResponse`
- `resume_conversation_stream(stream, registry, conversation_id, *, after=0) -> AsyncIterator[str] | None`
- `stop_conversation_stream(registry, stop, conversation_id) -> bool`

### `fastapi_resumable_stream.adapters`

- `from_langchain_events(events) -> AsyncIterator[str]`
- `from_crewai_stream(streaming) -> AsyncIterator[str]`
- `to_ag_ui_text_message(tokens, *, message_id=None, role="assistant") -> AsyncIterator[str]`
- `to_vercel_ui_message_stream(tokens, *, message_id=None, text_id=None) -> AsyncIterator[str]`
- `VERCEL_STREAM_HEADERS` — `{"x-vercel-ai-ui-message-stream": "v1"}`, pass as `extra_headers` to `streaming_response`/`start_conversation_stream`.

## License

[MIT](LICENSE) &copy; [Ofer Shapira](https://github.com/ofershap)
