🐛 Propagate `stream_item_type` through `include_router` for SSE / JSONL routes (#15401)

When an SSE route (or JSONL streaming route) is defined on an
`APIRouter` and merged onto a `FastAPI` app via `include_router`, the
merged route silently lost its `stream_item_type`. As a consequence the
emitted OpenAPI schema dropped the `contentSchema` describing the
streamed item and downstream tools (`datamodel-codegen`, etc.) could no
longer generate frame models from the spec.

Root cause: `APIRoute.__init__` only ran stream-item detection inside
the `isinstance(response_model, DefaultPlaceholder)` branch. The source
route's `__init__` collapses `self.response_model` to `None` after
detection, so when `APIRouter.include_router` re-instantiates the route
via `add_api_route(response_model=route.response_model, ...)` the new
init sees an explicit `None` and skips detection entirely.

Fix: also run detection when `response_model is None` on entry, and
restrict the "promote return annotation to response_model" fallback to
the original `DefaultPlaceholder` case so an explicit `response_model=
None` still disables response validation as documented.

Adds a regression test asserting both `stream_item_type` propagation and
the presence of `contentSchema` in the merged route's OpenAPI.

Co-Authored-By: Claude Code <[email protected]>

fastapi/routing.py

@@ -844,7 +844,17 @@ class APIRoute(routing.Route):
         self.path = path
         self.endpoint = endpoint
         self.stream_item_type: Any | None = None
-        if isinstance(response_model, DefaultPlaceholder):
+        # Run stream-item detection both when response_model is the
+        # DefaultPlaceholder (the original code path: user-defined route)
+        # and when it's an explicit ``None``. The ``None`` case fires when
+        # ``include_router`` re-instantiates a route from a source whose
+        # ``response_model`` was already collapsed to ``None`` during the
+        # first ``__init__``: without it, the merged route would silently
+        # drop ``stream_item_type`` and the OpenAPI ``contentSchema`` for
+        # the streaming endpoint (#15401). Explicit ``response_model=None``
+        # for streaming endpoints is also a documented pattern, so opting
+        # those into detection is intentional.
+        if isinstance(response_model, DefaultPlaceholder) or response_model is None:
             return_annotation = get_typed_return_annotation(endpoint)
             if lenient_issubclass(return_annotation, Response):
                 response_model = None
@@ -863,7 +873,12 @@ class APIRoute(routing.Route):
                     ) and not lenient_issubclass(stream_item, ServerSentEvent):
                         self.stream_item_type = stream_item
                     response_model = None
-                else:
+                elif isinstance(response_model, DefaultPlaceholder):
+                    # Only fall back to using the return annotation as the
+                    # response_model when the caller did not pass an
+                    # explicit value: an explicit ``None`` must stay
+                    # ``None`` (otherwise we'd reintroduce response
+                    # validation that the caller deliberately turned off).
                     response_model = return_annotation
         self.response_model = response_model
         self.summary = summary

tests/test_sse.py

@@ -316,3 +316,60 @@ def test_no_keepalive_when_fast(client: TestClient):
     assert response.status_code == 200
     # KEEPALIVE_COMMENT is ": ping\n\n".
     assert ": ping\n" not in response.text
+
+
+def test_stream_item_type_propagated_through_include_router():
+    """Regression test for #15401.
+
+    When an SSE route is defined on an ``APIRouter`` and merged onto a
+    ``FastAPI`` app via ``include_router``, the merged route must carry
+    the same ``stream_item_type`` as the source route, and the emitted
+    OpenAPI schema must include the ``contentSchema`` describing the
+    streamed item under
+    ``responses.200.content["text/event-stream"].itemSchema.properties.data``.
+
+    Before the fix, ``stream_item_type`` detection in ``APIRoute.__init__``
+    was gated on ``response_model`` being a ``DefaultPlaceholder``. The
+    source route's ``response_model`` was collapsed to ``None`` during
+    its first ``__init__``, so when ``include_router`` re-instantiated
+    the route with that ``None``, detection was skipped and the merged
+    route's ``stream_item_type`` stayed ``None``.
+    """
+
+    class Frame(BaseModel):
+        kind: str
+
+    # Case A — route registered directly on the app (control)
+    app_a = FastAPI()
+
+    @app_a.post("/s", response_class=EventSourceResponse)
+    async def direct() -> AsyncIterable[Frame]:
+        yield Frame(kind="x")
+
+    # Case B — route on a router, then ``include_router`` (regression case)
+    router_b = APIRouter()
+
+    @router_b.post("/s", response_class=EventSourceResponse)
+    async def via_router() -> AsyncIterable[Frame]:
+        yield Frame(kind="x")
+
+    app_b = FastAPI()
+    app_b.include_router(router_b)
+
+    # Both routes must carry the detected item type.
+    direct_route = app_a.routes[-1]
+    merged_route = app_b.routes[-1]
+    assert direct_route.stream_item_type is Frame  # type: ignore[union-attr]
+    assert merged_route.stream_item_type is Frame  # type: ignore[union-attr]
+
+    # And both must surface the contentSchema in the emitted OpenAPI.
+    def has_content_schema(spec: dict) -> bool:
+        sse = spec["paths"]["/s"]["post"]["responses"]["200"]["content"][
+            "text/event-stream"
+        ]
+        return "contentSchema" in sse.get("itemSchema", {}).get("properties", {}).get(
+            "data", {}
+        )
+
+    assert has_content_schema(app_a.openapi())
+    assert has_content_schema(app_b.openapi())