Skip to main content
FastAPI Docs

Search documentation

Type to search this documentation.

On this pageOverview

WebSockets

When defining WebSockets, you normally declare a parameter of type WebSocket and with it you can read data from the client and send data to it.

Read more about it in the FastAPI docs for WebSockets

It is provided directly by Starlette, but you can import it from fastapi:

from fastapi import WebSocket

fastapi.WebSocket

Section titled “fastapi.WebSocket”
WebSocket(scope, receive, send)

Bases: HTTPConnection[StateT]

Source code in starlette/websockets.py
def __init__(self, scope: Scope, receive: Receive, send: Send) -> None:
    super().__init__(scope)
    assert scope["type"] == "websocket"
    self._receive = receive
    self._send = send
    self.client_state = WebSocketState.CONNECTING
    self.application_state = WebSocketState.CONNECTING

scope instance-attribute

Section titled “scope instance-attribute”
scope = scope

app property

Section titled “app property”
app

url property

Section titled “url property”
url

base_url property

Section titled “base_url property”
base_url

headers property

Section titled “headers property”
headers

query_params property

Section titled “query_params property”
query_params

path_params property

Section titled “path_params property”
path_params

cookies property

Section titled “cookies property”
cookies

client property

Section titled “client property”
client

state property

Section titled “state property”
state

client_state instance-attribute

Section titled “client_state instance-attribute”
client_state = CONNECTING

application_state instance-attribute

Section titled “application_state instance-attribute”
application_state = CONNECTING

url_for

Section titled “url_for”
url_for(name, /, **path_params)
code
def url_for(self, name: str, /, **path_params: Any) -> URL:
    url_path_provider: Router | Starlette | None = self.scope.get("router") or self.scope.get("app")
    if url_path_provider is None:
        raise RuntimeError("The `url_for` method can only be used inside a Starlette application or with a router.")
    url_path = url_path_provider.url_path_for(name, **path_params)
    return url_path.make_absolute_url(base_url=self.base_url)

receive async

Section titled “receive async”
receive()

Receive ASGI websocket messages, ensuring valid state transitions.

Source code in starlette/websockets.py
async def receive(self) -> Message:
    """
    Receive ASGI websocket messages, ensuring valid state transitions.
    """
    if self.client_state == WebSocketState.CONNECTING:
        message = await self._receive()
        message_type = message["type"]
        if message_type != "websocket.connect":
            raise RuntimeError(f'Expected ASGI message "websocket.connect", but got {message_type!r}')
        self.client_state = WebSocketState.CONNECTED
        return message
    elif self.client_state == WebSocketState.CONNECTED:
        message = await self._receive()
        message_type = message["type"]
        if message_type not in {"websocket.receive", "websocket.disconnect"}:
            raise RuntimeError(
                f'Expected ASGI message "websocket.receive" or "websocket.disconnect", but got {message_type!r}'
            )
        if message_type == "websocket.disconnect":
            self.client_state = WebSocketState.DISCONNECTED
        return message
    else:
        raise RuntimeError('Cannot call "receive" once a disconnect message has been received.')

send async

Section titled “send async”
send(message)

Send ASGI websocket messages, ensuring valid state transitions.

Source code in starlette/websockets.py
async def send(self, message: Message) -> None:
    """
    Send ASGI websocket messages, ensuring valid state transitions.
    """
    if self.application_state == WebSocketState.CONNECTING:
        message_type = message["type"]
        if message_type not in {"websocket.accept", "websocket.close", "websocket.http.response.start"}:
            raise RuntimeError(
                'Expected ASGI message "websocket.accept", "websocket.close" or "websocket.http.response.start", '
                f"but got {message_type!r}"
            )
        if message_type == "websocket.close":
            self.application_state = WebSocketState.DISCONNECTED
        elif message_type == "websocket.http.response.start":
            self.application_state = WebSocketState.RESPONSE
        else:
            self.application_state = WebSocketState.CONNECTED
        await self._send(message)
    elif self.application_state == WebSocketState.CONNECTED:
        message_type = message["type"]
        if message_type not in {"websocket.send", "websocket.close"}:
            raise RuntimeError(
                f'Expected ASGI message "websocket.send" or "websocket.close", but got {message_type!r}'
            )
        if message_type == "websocket.close":
            self.application_state = WebSocketState.DISCONNECTED
        try:
            await self._send(message)
        except OSError:
            self.application_state = WebSocketState.DISCONNECTED
            raise WebSocketDisconnect(code=1006)
    elif self.application_state == WebSocketState.RESPONSE:
        message_type = message["type"]
        if message_type != "websocket.http.response.body":
            raise RuntimeError(f'Expected ASGI message "websocket.http.response.body", but got {message_type!r}')
        if not message.get("more_body", False):
            self.application_state = WebSocketState.DISCONNECTED
        await self._send(message)
    else:
        raise RuntimeError('Cannot call "send" once a close message has been sent.')

accept async

Section titled “accept async”
accept(subprotocol=None, headers=None)
code
async def accept(
    self,
    subprotocol: str | None = None,
    headers: Iterable[tuple[bytes, bytes]] | None = None,
) -> None:
    headers = headers or []

    if self.client_state == WebSocketState.CONNECTING:  # pragma: no branch
        # If we haven't yet seen the 'connect' message, then wait for it first.
        await self.receive()
    await self.send({"type": "websocket.accept", "subprotocol": subprotocol, "headers": headers})

receive_text async

Section titled “receive_text async”
receive_text()
code
async def receive_text(self) -> str:
    if self.application_state != WebSocketState.CONNECTED:
        raise RuntimeError('WebSocket is not connected. Need to call "accept" first.')
    message = await self.receive()
    self._raise_on_disconnect(message)
    return cast(str, message["text"])

receive_bytes async

Section titled “receive_bytes async”
receive_bytes()
code
async def receive_bytes(self) -> bytes:
    if self.application_state != WebSocketState.CONNECTED:
        raise RuntimeError('WebSocket is not connected. Need to call "accept" first.')
    message = await self.receive()
    self._raise_on_disconnect(message)
    return cast(bytes, message["bytes"])

receive_json async

Section titled “receive_json async”
receive_json(mode='text')
code
async def receive_json(self, mode: str = "text") -> Any:
    if mode not in {"text", "binary"}:
        raise RuntimeError('The "mode" argument should be "text" or "binary".')
    if self.application_state != WebSocketState.CONNECTED:
        raise RuntimeError('WebSocket is not connected. Need to call "accept" first.')
    message = await self.receive()
    self._raise_on_disconnect(message)

    if mode == "text":
        text = message["text"]
    else:
        text = message["bytes"].decode("utf-8")
    return json.loads(text)

iter_text async

Section titled “iter_text async”
iter_text()
code
async def iter_text(self) -> AsyncIterator[str]:
    try:
        while True:
            yield await self.receive_text()
    except WebSocketDisconnect:
        pass

iter_bytes async

Section titled “iter_bytes async”
iter_bytes()
code
async def iter_bytes(self) -> AsyncIterator[bytes]:
    try:
        while True:
            yield await self.receive_bytes()
    except WebSocketDisconnect:
        pass

iter_json async

Section titled “iter_json async”
iter_json()
code
async def iter_json(self) -> AsyncIterator[Any]:
    try:
        while True:
            yield await self.receive_json()
    except WebSocketDisconnect:
        pass

send_text async

Section titled “send_text async”
send_text(data)
code
async def send_text(self, data: str) -> None:
    await self.send({"type": "websocket.send", "text": data})

send_bytes async

Section titled “send_bytes async”
send_bytes(data)
code
async def send_bytes(self, data: bytes) -> None:
    await self.send({"type": "websocket.send", "bytes": data})

send_json async

Section titled “send_json async”
send_json(data, mode='text')
code
async def send_json(self, data: Any, mode: str = "text") -> None:
    if mode not in {"text", "binary"}:
        raise RuntimeError('The "mode" argument should be "text" or "binary".')
    text = json.dumps(data, separators=(",", ":"), ensure_ascii=False)
    if mode == "text":
        await self.send({"type": "websocket.send", "text": text})
    else:
        await self.send({"type": "websocket.send", "bytes": text.encode("utf-8")})

close async

Section titled “close async”
close(code=1000, reason=None)
code
async def close(self, code: int = 1000, reason: str | None = None) -> None:
    await self.send({"type": "websocket.close", "code": code, "reason": reason or ""})

WebSockets - additional classes

Section titled “WebSockets - additional classes”

Additional classes for handling WebSockets.

Provided directly by Starlette, but you can import them from fastapi:

from fastapi.websockets import WebSocketDisconnect, WebSocketState

fastapi.websockets.WebSocketDisconnect

Section titled “fastapi.websockets.WebSocketDisconnect”
WebSocketDisconnect(code=1000, reason=None)

Bases: Exception

Source code in starlette/websockets.py
def __init__(self, code: int = 1000, reason: str | None = None) -> None:
    self.code = code
    self.reason = reason or ""

code instance-attribute

Section titled “code instance-attribute”
code = code

reason instance-attribute

Section titled “reason instance-attribute”
reason = reason or ''

When a client disconnects, a WebSocketDisconnect exception is raised, you can catch it.

You can import it directly from fastapi:

from fastapi import WebSocketDisconnect

Read more about it in the FastAPI docs for WebSockets

fastapi.websockets.WebSocketState

Section titled “fastapi.websockets.WebSocketState”

Bases: Enum

CONNECTING class-attribute instance-attribute

Section titled “CONNECTING class-attribute instance-attribute”
CONNECTING = 0

CONNECTED class-attribute instance-attribute

Section titled “CONNECTED class-attribute instance-attribute”
CONNECTED = 1

DISCONNECTED class-attribute instance-attribute

Section titled “DISCONNECTED class-attribute instance-attribute”
DISCONNECTED = 2

RESPONSE class-attribute instance-attribute

Section titled “RESPONSE class-attribute instance-attribute”
RESPONSE = 3

WebSocketState is an enumeration of the possible states of a WebSocket connection.

Suggest an edit

Propose a replacement for this page. The site team reviews it before applying any changes.

Export
Documentation menu