Attach event data schema for sse endpoints

[BHznJNs]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from collections.abc import AsyncIterable
from fastapi import APIRouter
from fastapi.sse import EventSourceResponse
from pydantic import BaseModel

sse_router = APIRouter(tags=["stream"])

class ExampleEvent(BaseModel):
    message: str
    id: int

@sse_router.post(
    "/",
    response_class=EventSourceResponse,
)
async def sse_endpoint() -> AsyncIterable[ExampleEvent]:
    for i in range(10):
        yield ExampleEvent(message=f"Hello {i}", id=i)

Description

With the sse endpoint above, the generated openapi definition does not contain the yielded event type ExampleEvent, and it just contains a very general definition like

    "/api/events/": {
      "post": {
        "tags": [
          "stream"
        ],
        "summary": "Sse Endpoint",
        "operationId": "sse_endpoint",
        "responses": {
          "200": {
            "description": "Successful Response",
            "content": {
              "text/event-stream": {
                "itemSchema": {
                  "type": "object",
                  "properties": {
                    "data": {
                      "type": "string"
                    },
                    "event": {
                      "type": "string"
                    },
                    "id": {
                      "type": "string"
                    },
                    "retry": {
                      "type": "integer",
                      "minimum": 0
                    }
                  }
                }
              }
            }
          },
          "400": {
...

Operating System

Windows

Operating System Details

No response

FastAPI Version

0.135.1

Pydantic Version

2.12.5

Python Version

3.14.2

Additional Context

I also tried attach the event type via responses parameter like this:

@sse_router.post(
    "/",
    responses={ 200: {"model": ExampleEvent} },
    response_class=EventSourceResponse,
)
async def sse_endpoint() -> AsyncIterable[ExampleEvent]:
    for i in range(10):
        yield ExampleEvent(message=f"Hello {i}", id=i)

it actually included the ExampleEvent in the openapi definition, but it also includes an extra itemSchema field, which will cause some frontend code generator throws error (in my case, it's orval):

          "200": {
            "description": "Successful Response",
            "content": {
              "text/event-stream": {
                "schema": {
                  "$ref": "#/components/schemas/ExampleEvent"
                },
                "itemSchema": {
                  "type": "object",
                  "properties": {
                    "data": {
                      "type": "string"
                    },
                    "event": {
                      "type": "string"
                    },
                    "id": {
                      "type": "string"
                    },
                    "retry": {
                      "type": "integer",
                      "minimum": 0
                    }
                  }
                }
              }
            }
          },

[BHznJNs]

Update: I noticed that the itemSchema is a newly standard field of openapi spec, I wrongly thought that it's a fastapi's private field. Already openned issue at orval side: orval-labs/orval#3072

[alex-raw]

I hit the same thing and I believe the root cause is include_router(). Same endpoint, different schema depending on how it's registered:

from collections.abc import AsyncIterable

from fastapi import APIRouter, FastAPI
from fastapi.sse import EventSourceResponse
from pydantic import BaseModel


class ExampleEvent(BaseModel):
    message: str
    id: int

app = FastAPI()
sse_router = APIRouter()

@app.post("/direct", response_class=EventSourceResponse)
async def direct() -> AsyncIterable[ExampleEvent]:
    for i in range(10):
        yield ExampleEvent(message=f"Hello {i}", id=i)

@sse_router.post("/routed", response_class=EventSourceResponse)
async def routed() -> AsyncIterable[ExampleEvent]:
    for i in range(10):
        yield ExampleEvent(message=f"Hello {i}", id=i)

app.include_router(sse_router)

/direct generates the correct itemSchema -- data has contentMediaType and contentSchema with the $ref:

"itemSchema": {
  "type": "object",
  "properties": {
    "data": {
      "type": "string",
      "contentMediaType": "application/json",
      "contentSchema": { "$ref": "#/components/schemas/ExampleEvent" }
    },
    "event": { "type": "string" },
    "id": { "type": "string" },
    "retry": { "type": "integer", "minimum": 0 }
  },
  "required": ["data"]
}

/routed loses the item type -- contentMediaType, contentSchema, and required are all dropped:

"itemSchema": {
  "type": "object",
  "properties": {
    "data": { "type": "string" },
    "event": { "type": "string" },
    "id": { "type": "string" },
    "retry": { "type": "integer", "minimum": 0 }
  }
}

So the missing schema is likely not an orval issue -- it looks like FastAPI itself generates an incomplete itemSchema when the endpoint goes through a router. FastAPI 0.135.1, Python 3.13.

[tiangolo]

Thanks @alex-raw for the PR fixing it! It should now be available in FastAPI 0.140.8 🚀

[commonism]

c.f. #15328