aio-libs/aiohttp · error · RuntimeError

Cannot call .write() for websocket

Error message

Cannot call .write() for websocket

What it means

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.

Solutions

1. Replace ws.write(data) with ws.send_bytes(data) (for bytes) or ws.send_str(data) (for text).
2. In generic helpers, branch on isinstance(resp, WebSocketResponse) and use the appropriate send method.
3. Never call the inherited StreamResponse.write API on a WebSocketResponse.

Example fix

// before
await ws.write(b'chunk')  # raises

// after
await ws.send_bytes(b'chunk')
# or for text:
await ws.send_str('chunk')

if isinstance(resp, web.WebSocketResponse):
    await resp.send_bytes(data)
else:
    await resp.write(data)  # normal StreamResponse

def is_websocket_response(resp) -> bool:
    return isinstance(resp, web.WebSocketResponse)

try:
    await ws.write(data)
except RuntimeError as e:
    if 'Cannot call .write() for websocket' in str(e):
        await ws.send_bytes(data) if isinstance(data, (bytes, bytearray, memoryview)) else await ws.send_str(data)
    else:
        raise

When it happens

Trigger: 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.

Common situations: 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().

Related errors

• 1002
• Concurrent call to receive() is not allowed
• data argument must be byte-ish (%r)
• Response has not been started
• WebSocket connection is closed.

AI-assisted analysis of aio-libs/aiohttp@d041d4d0fd (2026-08-11). Data as JSON: /api/errors/9cf553a76ce07e40. Report an issue: GitHub.

Appendix: source

Thrown at aiohttp/web_ws.py:742

        self: "WebSocketResponse[_DecodeText]",
        *,
        loads: JSONDecoder | Callable[[bytes], Any] = ...,
        timeout: float | None = None,
    ) -> Any: ...

    async def receive_json(
        self,
        *,
        loads: JSONDecoder | Callable[[bytes], Any] = json.loads,
        timeout: float | None = None,
    ) -> Any:
        data = await self.receive_str(timeout=timeout)
        return loads(data)  # type: ignore[arg-type]

    async def write(
        self, data: Union[bytes, bytearray, "memoryview[int]", "memoryview[bytes]"]
    ) -> None:
        raise RuntimeError("Cannot call .write() for websocket")

    def __aiter__(self) -> Self:
        return self

    @overload
    async def __anext__(
        self: "WebSocketResponse[Literal[True]]",
    ) -> WSMessageDecodeText: ...

    @overload
    async def __anext__(
        self: "WebSocketResponse[Literal[False]]",
    ) -> WSMessageNoDecodeText: ...

    @overload
    async def __anext__(
        self: "WebSocketResponse[_DecodeText]",
    ) -> WSMessageDecodeText | WSMessageNoDecodeText: ...

View on GitHub (pinned to d041d4d0fd)