From 8efea2456763da287d42e8de44431d623ebea6ef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebasti=C3=A1n=20Ram=C3=ADrez?= Date: Sun, 1 Mar 2026 09:05:14 +0100 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Add=20support=20for=20Server=20Sent?= =?UTF-8?q?=20Events?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- fastapi/openapi/utils.py | 21 ++++ fastapi/responses.py | 1 + fastapi/routing.py | 180 +++++++++++++++++++++++++------ fastapi/sse.py | 222 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 389 insertions(+), 35 deletions(-) create mode 100644 fastapi/sse.py diff --git a/fastapi/openapi/utils.py b/fastapi/openapi/utils.py index 3ddc0c14a9..828442559b 100644 --- a/fastapi/openapi/utils.py +++ b/fastapi/openapi/utils.py @@ -29,6 +29,7 @@ from fastapi.openapi.constants import METHODS_WITH_BODY, REF_PREFIX from fastapi.openapi.models import OpenAPI from fastapi.params import Body, ParamTypes from fastapi.responses import Response +from fastapi.sse import _SSE_EVENT_SCHEMA from fastapi.types import ModelNameMap from fastapi.utils import ( deep_dict_update, @@ -372,6 +373,26 @@ def get_openapi_path( operation.setdefault("responses", {}).setdefault( status_code, {} ).setdefault("content", {})["application/jsonl"] = jsonl_content + elif route.is_sse_stream: + sse_content: dict[str, Any] = {} + item_schema = copy.deepcopy(_SSE_EVENT_SCHEMA) + if route.stream_item_field: + content_schema = get_schema_from_model_field( + field=route.stream_item_field, + model_name_map=model_name_map, + field_mapping=field_mapping, + separate_input_output_schemas=separate_input_output_schemas, + ) + item_schema["required"] = ["data"] + item_schema["properties"]["data"] = { + "type": "string", + "contentMediaType": "application/json", + "contentSchema": content_schema, + } + sse_content["itemSchema"] = item_schema + operation.setdefault("responses", {}).setdefault( + status_code, {} + ).setdefault("content", {})["text/event-stream"] = sse_content elif route_response_media_type: response_schema = {"type": "string"} if lenient_issubclass(current_response_class, JSONResponse): diff --git a/fastapi/responses.py b/fastapi/responses.py index 5b1154c046..554b0952b0 100644 --- a/fastapi/responses.py +++ b/fastapi/responses.py @@ -1,6 +1,7 @@ from typing import Any from fastapi.exceptions import FastAPIDeprecationWarning +from fastapi.sse import EventSourceResponse as EventSourceResponse # noqa from starlette.responses import FileResponse as FileResponse # noqa from starlette.responses import HTMLResponse as HTMLResponse # noqa from starlette.responses import JSONResponse as JSONResponse # noqa diff --git a/fastapi/routing.py b/fastapi/routing.py index f00cd2ca75..a52271690f 100644 --- a/fastapi/routing.py +++ b/fastapi/routing.py @@ -56,6 +56,13 @@ from fastapi.exceptions import ( ResponseValidationError, WebSocketRequestValidationError, ) +from fastapi.sse import ( + _PING_INTERVAL, + KEEPALIVE_COMMENT, + EventSourceResponse, + ServerSentEvent, + format_sse_event, +) from fastapi.types import DecoratedCallable, IncEx from fastapi.utils import ( create_model_field, @@ -66,7 +73,7 @@ from fastapi.utils import ( from starlette import routing from starlette._exception_handler import wrap_app_handling_exceptions from starlette._utils import is_async_callable -from starlette.concurrency import run_in_threadpool +from starlette.concurrency import iterate_in_threadpool, run_in_threadpool from starlette.exceptions import HTTPException from starlette.requests import Request from starlette.responses import JSONResponse, Response, StreamingResponse @@ -361,6 +368,7 @@ def get_request_handler( actual_response_class: type[Response] = response_class.value else: actual_response_class = response_class + is_sse_stream = lenient_issubclass(actual_response_class, EventSourceResponse) if isinstance(strict_content_type, DefaultPlaceholder): actual_strict_content_type: bool = strict_content_type.value else: @@ -452,35 +460,125 @@ def get_request_handler( errors = solved_result.errors assert dependant.call # For types if not errors: - if is_json_stream: - # Generator endpoint: stream as JSONL + # Shared serializer for stream items (JSONL and SSE). + # Validates against stream_item_field when set, then + # serializes to JSON bytes. + def _serialize_data(data: Any) -> bytes: + if stream_item_field: + value, errors_ = stream_item_field.validate( + data, {}, loc=("response",) + ) + if errors_: + ctx = endpoint_ctx or EndpointContext() + raise ResponseValidationError( + errors=errors_, + body=data, + endpoint_ctx=ctx, + ) + return stream_item_field.serialize_json( + value, + include=response_model_include, + exclude=response_model_exclude, + by_alias=response_model_by_alias, + exclude_unset=response_model_exclude_unset, + exclude_defaults=response_model_exclude_defaults, + exclude_none=response_model_exclude_none, + ) + else: + data = jsonable_encoder(data) + return json.dumps(data).encode("utf-8") + + if is_sse_stream: + # Generator endpoint: stream as Server-Sent Events gen = dependant.call(**solved_result.values) - def _serialize_item(item: Any) -> bytes: - if stream_item_field: - value, errors = stream_item_field.validate( - item, {}, loc=("response",) - ) - if errors: - ctx = endpoint_ctx or EndpointContext() - raise ResponseValidationError( - errors=errors, - body=item, - endpoint_ctx=ctx, - ) - line = stream_item_field.serialize_json( - value, - include=response_model_include, - exclude=response_model_exclude, - by_alias=response_model_by_alias, - exclude_unset=response_model_exclude_unset, - exclude_defaults=response_model_exclude_defaults, - exclude_none=response_model_exclude_none, + def _serialize_sse_item(item: Any) -> bytes: + if isinstance(item, ServerSentEvent): + # User controls the event structure. + # Serialize the data payload if present. + # For ServerSentEvent items we skip stream_item_field + # validation (the user may mix types intentionally). + if item.raw_data is not None: + data_str: str | None = item.raw_data + elif item.data is not None: + if hasattr(item.data, "model_dump_json"): + data_str = item.data.model_dump_json() + else: + data_str = json.dumps(jsonable_encoder(item.data)) + else: + data_str = None + return format_sse_event( + data_str=data_str, + event=item.event, + id=item.id, + retry=item.retry, + comment=item.comment, ) - return line + b"\n" else: - data = jsonable_encoder(item) - return json.dumps(data).encode("utf-8") + b"\n" + # Plain object: validate + serialize via + # stream_item_field (if set) and wrap in data field + return format_sse_event( + data_str=_serialize_data(item).decode("utf-8") + ) + + if dependant.is_async_gen_callable: + sse_aiter: AsyncIterator[Any] = gen.__aiter__() + else: + sse_aiter = iterate_in_threadpool(gen) + + async def _async_stream_sse() -> AsyncIterator[bytes]: + # Use a memory stream to decouple generator iteration + # from the keepalive timer. A producer task pulls items + # from the generator independently, so + # `anyio.fail_after` never wraps the generator's + # `__anext__` directly - avoiding CancelledError that + # would finalize the generator and also working for sync + # generators running in a thread pool. + send_stream, receive_stream = anyio.create_memory_object_stream[ + bytes + ](max_buffer_size=1) + + async def _producer() -> None: + async with send_stream: + async for raw_item in sse_aiter: + await send_stream.send(_serialize_sse_item(raw_item)) + + async with anyio.create_task_group() as tg: + tg.start_soon(_producer) + async with receive_stream: + try: + while True: + try: + with anyio.fail_after(_PING_INTERVAL): + data = await receive_stream.receive() + yield data + # To allow for cancellation to trigger + # Ref: https://github.com/fastapi/fastapi/issues/14680 + await anyio.sleep(0) + except TimeoutError: + yield KEEPALIVE_COMMENT + except anyio.EndOfStream: + pass + + sse_stream_content: AsyncIterator[bytes] | Iterator[bytes] = ( + _async_stream_sse() + ) + + response = StreamingResponse( + sse_stream_content, + media_type="text/event-stream", + background=solved_result.background_tasks, + ) + response.headers["Cache-Control"] = "no-cache" + # For Nginx proxies to not buffer server sent events + response.headers["X-Accel-Buffering"] = "no" + response.headers.raw.extend(solved_result.response.headers.raw) + elif is_json_stream: + # Generator endpoint: stream as JSONL + gen = dependant.call(**solved_result.values) + + def _serialize_item(item: Any) -> bytes: + return _serialize_data(item) + b"\n" if dependant.is_async_gen_callable: @@ -491,7 +589,7 @@ def get_request_handler( # Ref: https://github.com/fastapi/fastapi/issues/14680 await anyio.sleep(0) - stream_content: AsyncIterator[bytes] | Iterator[bytes] = ( + jsonl_stream_content: AsyncIterator[bytes] | Iterator[bytes] = ( _async_stream_jsonl() ) else: @@ -500,10 +598,10 @@ def get_request_handler( for item in gen: yield _serialize_item(item) - stream_content = _sync_stream_jsonl() + jsonl_stream_content = _sync_stream_jsonl() response = StreamingResponse( - stream_content, + jsonl_stream_content, media_type="application/jsonl", background=solved_result.background_tasks, ) @@ -709,9 +807,16 @@ class APIRoute(routing.Route): else: stream_item = get_stream_item_type(return_annotation) if stream_item is not None: - # Only extract item type for JSONL streaming when no - # explicit response_class (e.g. StreamingResponse) was set - if isinstance(response_class, DefaultPlaceholder): + # Extract item type for JSONL or SSE streaming when + # response_class is DefaultPlaceholder (JSONL) or + # EventSourceResponse (SSE). + # ServerSentEvent is excluded: it's a transport + # wrapper, not a data model, so it shouldn't feed + # into validation or OpenAPI schema generation. + if ( + isinstance(response_class, DefaultPlaceholder) + or lenient_issubclass(response_class, EventSourceResponse) + ) and not lenient_issubclass(stream_item, ServerSentEvent): self.stream_item_type = stream_item response_model = None else: @@ -814,11 +919,16 @@ class APIRoute(routing.Route): name=self.unique_id, embed_body_fields=self._embed_body_fields, ) - # Detect generator endpoints that should stream as JSONL - # (only when no explicit response_class like StreamingResponse is set) - self.is_json_stream = isinstance(response_class, DefaultPlaceholder) and ( + # Detect generator endpoints that should stream as JSONL or SSE + is_generator = ( self.dependant.is_async_gen_callable or self.dependant.is_gen_callable ) + self.is_sse_stream = is_generator and lenient_issubclass( + response_class, EventSourceResponse + ) + self.is_json_stream = is_generator and isinstance( + response_class, DefaultPlaceholder + ) self.app = request_response(self.get_route_handler()) def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]: diff --git a/fastapi/sse.py b/fastapi/sse.py new file mode 100644 index 0000000000..bd14713617 --- /dev/null +++ b/fastapi/sse.py @@ -0,0 +1,222 @@ +from typing import Annotated, Any + +from annotated_doc import Doc +from pydantic import AfterValidator, BaseModel, Field, model_validator +from starlette.responses import StreamingResponse + +# Canonical SSE event schema matching the OpenAPI 3.2 spec +# (Section 4.14.4 "Special Considerations for Server-Sent Events") +_SSE_EVENT_SCHEMA: dict[str, Any] = { + "type": "object", + "properties": { + "data": {"type": "string"}, + "event": {"type": "string"}, + "id": {"type": "string"}, + "retry": {"type": "integer", "minimum": 0}, + }, +} + + +class EventSourceResponse(StreamingResponse): + """Streaming response with `text/event-stream` media type. + + Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield` + to enable Server Sent Events (SSE) responses. + + Works with **any HTTP method** (`GET`, `POST`, etc.), which makes it compatible + with protocols like MCP that stream SSE over `POST`. + + The actual encoding logic lives in the FastAPI routing layer. This class + serves mainly as a marker and sets the correct `Content-Type`. + """ + + media_type = "text/event-stream" + + +def _check_id_no_null(v: int | str | None) -> int | str | None: + if v is not None and "\0" in str(v): + raise ValueError("SSE 'id' must not contain null characters") + return v + + +class ServerSentEvent(BaseModel): + """Represents a single Server-Sent Event. + + When `yield`ed from a *path operation function* that uses + `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded + into the [SSE wire format](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream) + (`text/event-stream`). + + If you yield a plain object (dict, Pydantic model, etc.) instead, it is + automatically JSON-encoded and sent as the `data:` field. + + All `data` values **including plain strings** are JSON-serialized. + + For example, `data="hello"` produces `data: "hello"` on the wire (with + quotes). + """ + + data: Annotated[ + Any, + Doc( + """ + The event payload. + + Can be any JSON-serializable value: a Pydantic model, dict, list, + string, number, etc. It is **always** serialized to JSON: strings + are quoted (`"hello"` becomes `data: "hello"` on the wire). + + Mutually exclusive with `raw_data`. + """ + ), + ] = None + raw_data: Annotated[ + str | None, + Doc( + """ + Raw string to send as the `data:` field **without** JSON encoding. + + Use this when you need to send pre-formatted text, HTML fragments, + CSV lines, or any non-JSON payload. The string is placed directly + into the `data:` field as-is. + + Mutually exclusive with `data`. + """ + ), + ] = None + event: Annotated[ + str | None, + Doc( + """ + Optional event type name. + + Maps to `addEventListener(event, ...)` on the browser. When omitted, + the browser dispatches on the generic `message` event. + """ + ), + ] = None + id: Annotated[ + int | str | None, + AfterValidator(_check_id_no_null), + Doc( + """ + Optional event ID. + + The browser sends this value back as the `Last-Event-ID` header on + automatic reconnection. **Must not contain null (`\\0`) characters.** + """ + ), + ] = None + retry: Annotated[ + int | None, + Field(ge=0), + Doc( + """ + Optional reconnection time in **milliseconds**. + + Tells the browser how long to wait before reconnecting after the + connection is lost. Must be a non-negative integer. + """ + ), + ] = None + comment: Annotated[ + str | None, + Doc( + """ + Optional comment line(s). + + Comment lines start with `:` in the SSE wire format and are ignored by + `EventSource` clients. Useful for keep-alive pings to prevent + proxy/load-balancer timeouts. + """ + ), + ] = None + + @model_validator(mode="after") + def _check_data_exclusive(self) -> "ServerSentEvent": + if self.data is not None and self.raw_data is not None: + raise ValueError( + "Cannot set both 'data' and 'raw_data' on the same " + "ServerSentEvent. Use 'data' for JSON-serialized payloads " + "or 'raw_data' for pre-formatted strings." + ) + return self + + +def format_sse_event( + *, + data_str: Annotated[ + str | None, + Doc( + """ + Pre-serialized data string to use as the `data:` field. + """ + ), + ] = None, + event: Annotated[ + str | None, + Doc( + """ + Optional event type name (`event:` field). + """ + ), + ] = None, + id: Annotated[ + int | str | None, + Doc( + """ + Optional event ID (`id:` field). + """ + ), + ] = None, + retry: Annotated[ + int | None, + Doc( + """ + Optional reconnection time in milliseconds (`retry:` field). + """ + ), + ] = None, + comment: Annotated[ + str | None, + Doc( + """ + Optional comment line(s) (`:` prefix). + """ + ), + ] = None, +) -> bytes: + """Build SSE wire-format bytes from **pre-serialized** data. + + The result always ends with `\n\n` (the event terminator). + """ + lines: list[str] = [] + + if comment is not None: + for line in comment.splitlines(): + lines.append(f": {line}") + + if event is not None: + lines.append(f"event: {event}") + + if data_str is not None: + for line in data_str.splitlines(): + lines.append(f"data: {line}") + + if id is not None: + lines.append(f"id: {id}") + + if retry is not None: + lines.append(f"retry: {retry}") + + lines.append("") + lines.append("") + return "\n".join(lines).encode("utf-8") + + +# Keep-alive comment, per the SSE spec recommendation +KEEPALIVE_COMMENT = b": ping\n\n" + +# Seconds between keep-alive pings when a generator is idle. +# Private but importable so tests can monkeypatch it. +_PING_INTERVAL: float = 15.0