Metadata-Version: 2.4
Name: fastapi-resumable-stream
Version: 0.1.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.

## API

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

- `start(stream_id, producer) -> AsyncIterator[str]` — starts (or attaches to) a stream, yielding chunks from the beginning.
- `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) -> StreamingResponse` — wraps an async iterator with the right SSE headers.
- `start_conversation_stream(stream, registry, stop, conversation_id, stream_id, producer) -> StreamingResponse`
- `resume_conversation_stream(stream, registry, conversation_id, *, after=0) -> AsyncIterator[str] | None`
- `stop_conversation_stream(registry, stop, conversation_id) -> bool`

## License

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