{"record":{"id":"9cf553a76ce07e40","repo":"aio-libs/aiohttp","slug":"cannot-call-write-for-websocket","errorCode":null,"errorMessage":"Cannot call .write() for websocket","messagePattern":"Cannot call \\.write\\(\\) for websocket","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"aiohttp/web_ws.py","lineNumber":742,"sourceCode":"        self: \"WebSocketResponse[_DecodeText]\",\n        *,\n        loads: JSONDecoder | Callable[[bytes], Any] = ...,\n        timeout: float | None = None,\n    ) -> Any: ...\n\n    async def receive_json(\n        self,\n        *,\n        loads: JSONDecoder | Callable[[bytes], Any] = json.loads,\n        timeout: float | None = None,\n    ) -> Any:\n        data = await self.receive_str(timeout=timeout)\n        return loads(data)  # type: ignore[arg-type]\n\n    async def write(\n        self, data: Union[bytes, bytearray, \"memoryview[int]\", \"memoryview[bytes]\"]\n    ) -> None:\n        raise RuntimeError(\"Cannot call .write() for websocket\")\n\n    def __aiter__(self) -> Self:\n        return self\n\n    @overload\n    async def __anext__(\n        self: \"WebSocketResponse[Literal[True]]\",\n    ) -> WSMessageDecodeText: ...\n\n    @overload\n    async def __anext__(\n        self: \"WebSocketResponse[Literal[False]]\",\n    ) -> WSMessageNoDecodeText: ...\n\n    @overload\n    async def __anext__(\n        self: \"WebSocketResponse[_DecodeText]\",\n    ) -> WSMessageDecodeText | WSMessageNoDecodeText: ...","sourceCodeStart":724,"sourceCodeEnd":760,"githubUrl":"https://github.com/aio-libs/aiohttp/blob/d041d4d0fd48c3f0832084d33be16cf1c4835f85/aiohttp/web_ws.py#L724-L760","documentation":"WebSocketResponse overrides StreamResponse.write() to unconditionally raise RuntimeError. A WebSocket has no HTTP response body to stream — data goes out as framed messages via send_str/send_bytes/send_json. The override exists to give a clear error instead of letting inherited write() corrupt the connection by emitting raw bytes that the WS parser would reject.","triggerScenarios":"Calling `await ws.write(b'...')` directly; generic streaming code or middleware that writes to any StreamResponse without checking for the WebSocket subtype; a helper that accepts a response and calls write() on it.","commonSituations":"Reusing a chunked-stream helper for both HTTP and WebSocket responses; copy-paste from a normal handler into a WebSocket handler; middleware that calls response.write().","solutions":["Replace ws.write(data) with ws.send_bytes(data) (for bytes) or ws.send_str(data) (for text).","In generic helpers, branch on isinstance(resp, WebSocketResponse) and use the appropriate send method.","Never call the inherited StreamResponse.write API on a WebSocketResponse."],"exampleFix":"// before\nawait ws.write(b'chunk')  # raises\n\n// after\nawait ws.send_bytes(b'chunk')\n# or for text:\nawait ws.send_str('chunk')","handlingStrategy":"type-guard","validationCode":"if isinstance(resp, web.WebSocketResponse):\n    await resp.send_bytes(data)\nelse:\n    await resp.write(data)  # normal StreamResponse","typeGuard":"def is_websocket_response(resp) -> bool:\n    return isinstance(resp, web.WebSocketResponse)","tryCatchPattern":"try:\n    await ws.write(data)\nexcept RuntimeError as e:\n    if 'Cannot call .write() for websocket' in str(e):\n        await ws.send_bytes(data) if isinstance(data, (bytes, bytearray, memoryview)) else await ws.send_str(data)\n    else:\n        raise","preventionTips":["Never call StreamResponse.write() on a WebSocketResponse; use send_str/send_bytes/send_json.","Make generic streaming helpers branch on isinstance(resp, WebSocketResponse).","In tests, exercise both HTTP and WebSocket paths through shared helpers."],"tags":["websocket","send","api-misuse","server"],"backgroundTag":null,"analyzedSha":"d041d4d0fd48c3f0832084d33be16cf1c4835f85","analyzedAt":"2026-08-11T20:44:15.550Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}