aio-libs/aiohttp · error · RuntimeError

Cannot call .write() for websocket

Error message

Cannot call .write() for websocket

What it means

Raised unconditionally by WebSocketResponse.write() (aiohttp/web_ws.py:739-742). WebSocketResponse inherits write() from StreamResponse but overrides it to always raise, because raw byte writes bypass WebSocket framing. You must use the frame-aware send_str/send_bytes/send_json methods which construct proper WebSocket frames via the internal writer.

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 c0ef574e29)

Solutions

1. Replace `await ws.write(data)` with `await ws.send_bytes(data)` for binary or `await ws.send_str(data)` for text.
2. Use send_json()/send_json_bytes() for JSON payloads.
3. Remove any generic write() call path that targets a WebSocketResponse.

Example fix

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

# after
await ws.send_bytes(b'chunk')

async def ws_send(ws, data):
    if isinstance(data, (bytes, bytearray, memoryview)):
        await ws.send_bytes(data)
    else:
        await ws.send_str(data)

def is_websocket_response(resp) -> bool:
    return type(resp).__name__ == "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 str(data).encode())
    else:
        raise

When it happens

Trigger: Calling `await ws.write(data)` on a WebSocketResponse at any time, prepared or not. The method body is just `raise RuntimeError(...)`.

Common situations: Treating a WebSocketResponse like a normal StreamResponse and calling write(); code copied from a streaming HTTP handler into a WS handler; a framework abstraction that calls write() generically on response objects.

Related errors

• Compress wbits must between 9 and 15, zlib does not support
• 1007
• data argument must be byte-ish (%r)
• Response has not been started
• Concurrent call to receive() is not allowed

AI-assisted analysis of aio-libs/aiohttp@c0ef574e29 (2026-08-04). Data as JSON: /data/errors/9cf553a76ce07e40.json. Report an issue: GitHub.