fastapi-streams

Decorate a FastAPI endpoint. Get typed streaming clients in TypeScript and Python.

Define your events with Pydantic and add @router.stream(). fastapi-streams turns that definition into an OpenAPI 3.2 stream contract and generated SDKs with named event values, typed async iteration, and shared SSE transport code.

FastAPI decorator → OpenAPI 3.2 → shared HTTP IR → TypeScript + Python clients

Why this exists

An endpoint labeled text/event-stream does not give consumers a typed SDK. They still need to know which JSON events arrive, how to parse a fragmented stream, and how to stop it cleanly.

OpenAPI 3.2 can describe each SSE event with itemSchema and its JSON payload with contentSchema. fastapi-streams connects that contract to both clients, so event models, endpoint parameters, and client types come from one definition. A CI check catches drift between the Python app, the contract, and generated code.

1. Decorate your FastAPI endpoint

Install from GitHub with the optional codegen extra:

uv add 'fastapi-streams[codegen] @ git+https://github.com/mathisarends/fastapi-streams.git'

Save this as app.py:

from collections.abc import AsyncIterator
from typing import Annotated, Literal

from fastapi import FastAPI
from pydantic import BaseModel, Field

from fastapi_streams import StreamRouter


class TokenDelta(BaseModel):
    type: Literal["token.delta"]
    text: str


class RunCompleted(BaseModel):
    type: Literal["run.completed"]
    run_id: str


type RunEvent = Annotated[TokenDelta | RunCompleted, Field(discriminator="type")]

router = StreamRouter(prefix="/runs", tags=["agentRuns"])


@router.stream("/{run_id}/events", operation_id="stream_run_events")
async def stream_run_events(run_id: str) -> AsyncIterator[RunEvent]:
    yield TokenDelta(type="token.delta", text="Hello")
    yield RunCompleted(type="run.completed", run_id=run_id)


app = FastAPI()
app.include_router(router, prefix="/v1")

The result is GET /v1/runs/{run_id}/events. StreamRouter extends APIRouter: normal Depends, security dependencies, validation, middleware, and nested routers keep working. FastAPI handles SSE framing, serialization, and heartbeats. Use methods=["POST"] for request-body-driven streams.

2. Export once, generate both clients

uv run fastapi-streams schema \
  --app app:app --output openapi.json

uv run fastapi-streams generate \
  --schema openapi.json --language typescript --output generated/typescript

uv run fastapi-streams generate \
  --schema openapi.json --language python --output generated/python

These commands import the app and read its contract; no running server is needed. The exporter validates OpenAPI 3.2 and the event schemas. Both generators consume one HTTP intermediate representation (IR), carrying operations, parameters, and event types into the generated SDKs.

3. Consume typed events

The RunEvent union produces RunEventType, with the same named values in both languages. The discriminator narrows each event to its matching model.

TypeScript — Fetch, AsyncIterable<RunEvent>, no runtime dependencies:

import { createClient, RunEventType } from "./generated/typescript/index.js";

const client = createClient({ baseUrl: "http://localhost:8000" });

for await (const event of client.agentRuns.streamRunEvents({ runId: "demo" })) {
  switch (event.type) {
    case RunEventType.TOKEN_DELTA:
      console.log(event.text); // TokenDelta
      break;
    case RunEventType.RUN_COMPLETED:
      console.log(event.run_id); // RunCompleted
      break;
    default: {
      const exhaustive: never = event;
      throw new Error(`Unexpected event: ${exhaustive}`);
    }
  }
}

Python — HTTPX, Pydantic models, and StrEnum values:

import asyncio
from typing import assert_never

from generated.python import Client, RunEventType


async def main() -> None:
    async with Client("http://localhost:8000") as client:
        async for event in client.agent_runs.stream_run_events(run_id="demo"):
            match event.type:
                case RunEventType.TOKEN_DELTA:
                    print(event.text)
                case RunEventType.RUN_COMPLETED:
                    print(event.run_id)
                case _:
                    assert_never(event)


asyncio.run(main())

Install httpx and pydantic in a Python consumer (uv add httpx pydantic). HTTPX remains the default; opt in to HTTPX2 at generation time with --python-http-client httpx2 and install httpx2 instead. See Python HTTP library selection. Both SDKs handle fragmented SSE frames, UTF-8 decoding, HTTP errors, and resource cleanup. TypeScript supports AbortSignal; Python supports task cancellation and explicit async-generator closure. See client usage.

4. Catch contract drift in CI

Commit the exported schema and generated clients. Then check the complete chain:

uv run fastapi-streams check \
  --app app:app --schema openapi.json \
  --language typescript --output generated/typescript

uv run fastapi-streams check \
  --app app:app --schema openapi.json \
  --language python --output generated/python

check fails if the app's contract changed or generated files are stale. It writes nothing and needs no running server. Pair it with TypeScript's compiler and Pyright to catch invalid parameters and event access in consumer code.

Codegen and CI reference →

Run the complete example

git clone https://github.com/mathisarends/fastapi-streams.git
cd fastapi-streams
uv sync --all-extras
npm ci
uv run uvicorn examples.agent_runs:app --reload

In another terminal:

# TypeScript
npx tsc -p tests/typescript/tsconfig.json
node tests/typescript/dist/examples/agent_runs/consume.js

# Python
uv run python -m examples.agent_runs.consume

The example includes GET and authenticated POST streams, both generated SDKs, and committed OpenAPI regression fixtures in tests/fixtures/. Verify all generated artifacts with uv run python scripts/codegen.py --check.

Documentation

v0.0.1 supports GET/POST JSON SSE streams. Ordinary REST endpoints stay in the contract for your existing REST SDK. See the supported scope.

Python 3.12+ · FastAPI 0.143.3+ · Pydantic v2 · MIT