fastapi_15588

{"status": "ok", "stdout": "./fastapi/.agents/skills/fastapi/references/streaming.md:16:To stream Server-Sent Events, use `response_class=EventSourceResponse` and `yield` items from the endpoint.\n./fastapi/.agents/skills/fastapi/references/streaming.md:24:from fastapi.sse import EventSourceResponse\n./fastapi/.agents/skills/fastapi/references/streaming.md:35:@app.get(\"/items/stream\", response_class=EventSourceResponse)\n./fastapi/.agents/skills/fastapi/references/streaming.md:47:from fastapi.sse import EventSourceResponse, ServerSentEvent\n./fastapi/.agents/skills/fastapi/references/streaming.md:52:@app.get(\"/events\", response_class=EventSourceResponse)\n./fastapi/.agents/skills/fastapi/SKILL.md:333:See [the streaming reference](references/streaming.md) for JSON Lines, Server-Sent Events (`EventSourceResponse`, `ServerSentEvent`), and byte streaming (`StreamingResponse`) patterns.\n./fastapi/routing.py:64:    EventSourceResponse,\n./fastapi/routing.py:376:    is_sse_stream = lenient_issubclass(actual_response_class, EventSourceResponse)\n./fastapi/routing.py:856:                    # EventSourceResponse (SSE).\n./fastapi/routing.py:862:                        or lenient_issubclass(response_class, EventSourceResponse)\n./fastapi/routing.py:971:            response_class, EventSourceResponse\n./fastapi/responses.py:5:from fastapi.sse import EventSourceResponse as EventSourceResponse  # noqa\n./fastapi/sse.py:20:class EventSourceResponse(StreamingResponse):\n./fastapi/sse.py:23:    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n./fastapi/sse.py:46:    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\n./tests/test_sse.py:8:from fastapi.responses import EventSourceResponse\n./tests/test_sse.py:29:@app.get(\"/items/stream\", response_class=EventSourceResponse)\n./tests/test_sse.py:35:@app.get(\"/items/stream-sync\", response_class=EventSourceResponse)\n./tests/test_sse.py:40:@app.get(\"/items/stream-no-anno

{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_validator\nfrom starlette.responses import StreamingResponse\n\n# Canonical SSE event schema matching the OpenAPI 3.2 spec\n# (Section 4.14.4 \"Special Considerations for Server-Sent Events\")\n_SSE_EVENT_SCHEMA: dict[str, Any] = {\n    \"type\": \"object\",\n    \"properties\": {\n        \"data\": {\"type\": \"string\"},\n        \"event\": {\"type\": \"string\"},\n        \"id\": {\"type\": \"string\"},\n        \"retry\": {\"type\": \"integer\", \"minimum\": 0},\n    },\n}\n\n\nclass EventSourceResponse(StreamingResponse):\n    \"\"\"Streaming response with `text/event-stream` media type.\n\n    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n    to enable Server Sent Events (SSE) responses.\n\n    Works with **any HTTP method** (`GET`, `POST`, etc.), which makes it compatible\n    with protocols like MCP that stream SSE over `POST`.\n\n    The actual encoding logic lives in the FastAPI routing layer. This class\n    serves mainly as a marker and sets the correct `Content-Type`.\n    \"\"\"\n\n    media_type = \"text/event-stream\"\n\n\ndef _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\n\n\nclass ServerSentEvent(BaseModel):\n    \"\"\"Represents a single Server-Sent Event.\n\n    When `yield`ed from a *path operation function* that uses\n    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\n    into the [SSE wire format](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream)\n    (`text/event-stream`).\n\n    If you yield a plain object (dict, Pydantic model, etc.) instead, it is\n    automatically JSON-encoded and sent as the `data:` field.\n\n    All `data` values **including plain string

{"status": "ok", "stdout": "        AfterValidator(_check_id_no_null),\n        Doc(\n            \"\"\"\n            Optional event ID.\n\n            The browser sends this value back as the `Last-Event-ID` header on\n            automatic reconnection. **Must not contain null (`\\\\0`) characters.**\n            \"\"\"\n        ),\n    ] = None\n    retry: Annotated[\n        int | None,\n        Field(ge=0),\n        Doc(\n            \"\"\"\n            Optional reconnection time in **milliseconds**.\n\n            Tells the browser how long to wait before reconnecting after the\n            connection is lost. Must be a non-negative integer.\n            \"\"\"\n        ),\n    ] = None\n    comment: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional comment line(s).\n\n            Comment lines start with `:` in the SSE wire format and are ignored by\n            `EventSource` clients. Useful for keep-alive pings to prevent\n            proxy/load-balancer timeouts.\n            \"\"\"\n        ),\n    ] = None\n\n    @model_validator(mode=\"after\")\n    def _check_data_exclusive(self) -> \"ServerSentEvent\":\n        if self.data is not None and self.raw_data is not None:\n            raise ValueError(\n                \"Cannot set both 'data' and 'raw_data' on the same \"\n                \"ServerSentEvent. Use 'data' for JSON-serialized payloads \"\n                \"or 'raw_data' for pre-formatted strings.\"\n            )\n        return self\n\n\ndef format_sse_event(\n    *,\n    data_str: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Pre-serialized data string to use as the `data:` field.\n            \"\"\"\n        ),\n    ] = None,\n    event: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional event type name (`event:` field).\n            \"\"\"\n        ),\n    ] = None,\n    id: Annotated[\n        str | None,\n        Doc(\n            \"

{"status": "ok", "stdout": "import asyncio\nimport time\nfrom collections.abc import AsyncIterable, Iterable\n\nimport fastapi.routing\nimport pytest\nfrom fastapi import APIRouter, FastAPI\nfrom fastapi.responses import EventSourceResponse\nfrom fastapi.sse import ServerSentEvent\nfrom fastapi.testclient import TestClient\nfrom pydantic import BaseModel\n\n\nclass Item(BaseModel):\n    name: str\n    description: str | None = None\n\n\nitems = [\n    Item(name=\"Plumbus\", description=\"A multi-purpose household device.\"),\n    Item(name=\"Portal Gun\", description=\"A portal opening device.\"),\n    Item(name=\"Meeseeks Box\", description=\"A box that summons a Meeseeks.\"),\n]\n\n\napp = FastAPI()\n\n\n@app.get(\"/items/stream\", response_class=EventSourceResponse)\nasync def sse_items() -> AsyncIterable[Item]:\n    for item in items:\n        yield item\n\n\n@app.get(\"/items/stream-sync\", response_class=EventSourceResponse)\ndef sse_items_sync() -> Iterable[Item]:\n    yield from items\n\n\n@app.get(\"/items/stream-no-annotation\", response_class=EventSourceResponse)\nasync def sse_items_no_annotation():\n    for item in items:\n        yield item\n\n\n@app.get(\"/items/stream-sync-no-annotation\", response_class=EventSourceResponse)\ndef sse_items_sync_no_annotation():\n    yield from items\n\n\n@app.get(\"/items/stream-dict\", response_class=EventSourceResponse)\nasync def sse_items_dict():\n    for item in items:\n        yield {\"name\": item.name, \"description\": item.description}\n\n\n@app.get(\"/items/stream-sse-event\", response_class=EventSourceResponse)\nasync def sse_items_event():\n    yield ServerSentEvent(data=\"hello\", event=\"greeting\", id=\"1\")\n    yield ServerSentEvent(data={\"key\": \"value\"}, event=\"json-data\", id=\"2\")\n    yield ServerSentEvent(comment=\"just a comment\")\n    yield ServerSentEvent(data=\"retry-test\", retry=5000)\n\n\n@app.get(\"/items/stream-mixed\", response_class=EventSourceResponse)\nasync def sse_items_mix

{"status": "ok", "stdout": "\ndef test_sync_generator_no_annotation(client: TestClient):\n    response = client.get(\"/items/stream-sync-no-annotation\")\n    assert response.status_code == 200\n    assert response.headers[\"content-type\"] == \"text/event-stream; charset=utf-8\"\n\n    data_lines = [\n        line for line in response.text.strip().split(\"\\n\") if line.startswith(\"data: \")\n    ]\n    assert len(data_lines) == 3\n\n\ndef test_dict_items(client: TestClient):\n    response = client.get(\"/items/stream-dict\")\n    assert response.status_code == 200\n    data_lines = [\n        line for line in response.text.strip().split(\"\\n\") if line.startswith(\"data: \")\n    ]\n    assert len(data_lines) == 3\n    assert '\"name\"' in data_lines[0]\n\n\ndef test_post_method_sse(client: TestClient):\n    \"\"\"SSE should work with POST (needed for MCP compatibility).\"\"\"\n    response = client.post(\"/items/stream-post\")\n    assert response.status_code == 200\n    assert response.headers[\"content-type\"] == \"text/event-stream; charset=utf-8\"\n    data_lines = [\n        line for line in response.text.strip().split(\"\\n\") if line.startswith(\"data: \")\n    ]\n    assert len(data_lines) == 3\n\n\ndef test_sse_events_with_fields(client: TestClient):\n    response = client.get(\"/items/stream-sse-event\")\n    assert response.status_code == 200\n    text = response.text\n\n    assert \"event: greeting\\n\" in text\n    assert 'data: \"hello\"\\n' in text\n    assert \"id: 1\\n\" in text\n\n    assert \"event: json-data\\n\" in text\n    assert \"id: 2\\n\" in text\n    assert 'data: {\"key\": \"value\"}\\n' in text\n\n    assert \": just a comment\\n\" in text\n\n    assert \"retry: 5000\\n\" in text\n    assert 'data: \"retry-test\"\\n' in text\n\n\ndef test_mixed_plain_and_sse_events(client: TestClient):\n    response = client.get(\"/items/stream-mixed\")\n    assert response.status_code == 200\n    text = response.text\n\n    assert \"event: special

{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.13.14, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: enabled, timer_resolution: 1.0ns)\nrootdir: /workspace\nconfigfile: pytest.ini (WARNING: ignoring pytest config in pyproject.toml!)\nplugins: anyio-4.15.1, sugar-1.1.1, inline-snapshot-0.35.4, timeout-2.4.0, xdist-3.8.0, codspeed-5.0.3, cov-7.1.0\ncollected 18 items\n\ntests/test_sse.py ..................                                     [100%]\n\n=============================== warnings summary ===============================\n../../../../../../kaggle/tmp/envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53\n  /kaggle/tmp/envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.\n    _PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]\n\n-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html\n======================== 18 passed, 1 warning in 1.25s =========================\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "     1\tfrom typing import Annotated, Any\n     2\t\n     3\tfrom annotated_doc import Doc\n     4\tfrom pydantic import AfterValidator, BaseModel, Field, model_validator\n     5\tfrom starlette.responses import StreamingResponse\n     6\t\n     7\t# Canonical SSE event schema matching the OpenAPI 3.2 spec\n     8\t# (Section 4.14.4 \"Special Considerations for Server-Sent Events\")\n     9\t_SSE_EVENT_SCHEMA: dict[str, Any] = {\n    10\t    \"type\": \"object\",\n    11\t    \"properties\": {\n    12\t        \"data\": {\"type\": \"string\"},\n    13\t        \"event\": {\"type\": \"string\"},\n    14\t        \"id\": {\"type\": \"string\"},\n    15\t        \"retry\": {\"type\": \"integer\", \"minimum\": 0},\n    16\t    },\n    17\t}\n    18\t\n    19\t\n    20\tclass EventSourceResponse(StreamingResponse):\n    21\t    \"\"\"Streaming response with `text/event-stream` media type.\n    22\t\n    23\t    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n    24\t    to enable Server Sent Events (SSE) responses.\n    25\t\n    26\t    Works with **any HTTP method** (`GET`, `POST`, etc.), which makes it compatible\n    27\t    with protocols like MCP that stream SSE over `POST`.\n    28\t\n    29\t    The actual encoding logic lives in the FastAPI routing layer. This class\n    30\t    serves mainly as a marker and sets the correct `Content-Type`.\n    31\t    \"\"\"\n    32\t\n    33\t    media_type = \"text/event-stream\"\n    34\t\n    35\t\n    36\tdef _check_id_no_null(v: str | None) -> str | None:\n    37\t    if v is not None and \"\\0\" in v:\n    38\t        raise ValueError(\"SSE 'id' must not contain null characters\")\n    39\t    return v\n    40\t\n    41\t\n    42\tclass ServerSentEvent(BaseModel):\n    43\t    \"\"\"Represents a single Server-Sent Event.\n    44\t\n    45\t    When `yield`ed from a *path operation function* that uses\n    46\t    `response_class=EventSourceResponse`, each 

{"status": "ok", "stdout": "            proxy/load-balancer timeouts.\n            \"\"\"\n        ),\n    ] = None\n\n    @model_validator(mode=\"after\")\n    def _check_data_exclusive(self) -> \"ServerSentEvent\":\n        if self.data is not None and self.raw_data is not None:\n            raise ValueError(\n                \"Cannot set both 'data' and 'raw_data' on the same \"\n                \"ServerSentEvent. Use 'data' for JSON-serialized payloads \"\n                \"or 'raw_data' for pre-formatted strings.\"\n            )\n        return self\n\n\ndef format_sse_event(\n    *,\n    data_str: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Pre-serialized data string to use as the `data:` field.\n            \"\"\"\n        ),\n    ] = None,\n    event: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional event type name (`event:` field).\n            \"\"\"\n        ),\n    ] = None,\n    id: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional event ID (`id:` field).\n            \"\"\"\n        ),\n    ] = None,\n    retry: Annotated[\n        int | None,\n        Doc(\n            \"\"\"\n            Optional reconnection time in milliseconds (`retry:` field).\n            \"\"\"\n        ),\n    ] = None,\n    comment: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional comment line(s) (`:` prefix).\n            \"\"\"\n        ),\n    ] = None,\n) -> bytes:\n    \"\"\"Build SSE wire-format bytes from **pre-serialized** data.\n\n    The result always ends with `\\n\\n` (the event terminator).\n    \"\"\"\n    lines: list[str] = []\n\n    if comment is not None:\n        for line in comment.splitlines():\n            lines.append(f\": {line}\")\n\n    if event is not None:\n        lines.append(f\"event: {event}\")\n\n    if data_str is not None:\n        for line in data_str.splitlines():\n            lines.append(f

{"status": "ok", "filepath": "fastapi/sse.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -33,9 +33,17 @@\n     media_type = \"text/event-stream\"\n \n \n+def _check_sse_field_no_newline(v: str | None) -> str | None:\n+    if v is not None and \"\\n\" in v:\n+        raise ValueError(\"SSE field must not contain newline characters\")\n+    return v\n+\n+\n def _check_id_no_null(v: str | None) -> str | None:\n     if v is not None and \"\\0\" in v:\n         raise ValueError(\"SSE 'id' must not contain null characters\")\n+    if v is not None and \"\\n\" in v:\n+        raise ValueError(\"SSE 'id' must not contain newline characters\")\n     return v\n \n \n", "is_truncated": false}

{"status": "ok", "filepath": "fastapi/sse.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -94,6 +94,7 @@\n     ] = None\n     event: Annotated[\n         str | None,\n+        AfterValidator(_check_sse_field_no_newline),\n         Doc(\n             \"\"\"\n             Optional event type name.\n", "is_truncated": false}

{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.13.14, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: enabled, timer_resolution: 1.0ns)\nrootdir: /workspace\nconfigfile: pytest.ini (WARNING: ignoring pytest config in pyproject.toml!)\nplugins: anyio-4.15.1, sugar-1.1.1, inline-snapshot-0.35.4, timeout-2.4.0, xdist-3.8.0, codspeed-5.0.3, cov-7.1.0\ncollected 18 items\n\ntests/test_sse.py ..................                                     [100%]\n\n=============================== warnings summary ===============================\n../../../../../../kaggle/tmp/envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53\n  /kaggle/tmp/envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.\n    _PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]\n\n-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html\n======================== 18 passed, 1 warning in 1.34s =========================\n", "stderr": "", "exit_code": 0}

{"status": "ok", "filepath": "tests/test_sse_validation.py", "size": 831}

{"status": "ok", "stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.13.14, pytest-9.1.1, pluggy-1.6.0\ncodspeed: 5.0.3 (disabled, mode: walltime, callgraph: enabled, timer_resolution: 1.0ns)\nrootdir: /workspace\nconfigfile: pytest.ini (WARNING: ignoring pytest config in pyproject.toml!)\nplugins: anyio-4.15.1, sugar-1.1.1, inline-snapshot-0.35.4, timeout-2.4.0, xdist-3.8.0, codspeed-5.0.3, cov-7.1.0\ncollected 3 items\n\ntests/test_sse_validation.py ...                                         [100%]\n\n============================== 3 passed in 0.41s ===============================\n", "stderr": "", "exit_code": 0}

{"status": "ok", "patch_size": 1981, "files_changed": 2}