# 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](/guides/advanced-user-guide-websockets)

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

```
from fastapi import WebSocket
```

:::callout{intent="tip"}
When you want to define dependencies that should be compatible with both HTTP and WebSockets, you can define a parameter that takes an `HTTPConnection` instead of a `Request` or a `WebSocket`.
:::

## fastapi.WebSocket [¶](#fastapiwebsocket-)

```
WebSocket(scope, receive, send)
```

Bases: `HTTPConnection[StateT]`

:::accordion{title="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` [¶](#scope-instance-attribute-)

```
scope = scope
```

### app `property` [¶](#app-property-)

```
app
```

### url `property` [¶](#url-property-)

```
url
```

### base\_url `property` [¶](#baseurl-property-)

```
base_url
```

### headers `property` [¶](#headers-property-)

```
headers
```

### query\_params `property` [¶](#queryparams-property-)

```
query_params
```

### path\_params `property` [¶](#pathparams-property-)

```
path_params
```

### cookies `property` [¶](#cookies-property-)

```
cookies
```

### client `property` [¶](#client-property-)

```
client
```

### state `property` [¶](#state-property-)

```
state
```

### client\_state `instance-attribute` [¶](#clientstate-instance-attribute-)

```
client_state = CONNECTING
```

### application\_state `instance-attribute` [¶](#applicationstate-instance-attribute-)

```
application_state = CONNECTING
```

### url\_for [¶](#urlfor-)

```
url_for(name, /, **path_params)
```

```title="Source code in starlette/requests.py"
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` [¶](#receive-async-)

```
receive()
```

Receive ASGI websocket messages, ensuring valid state transitions.

:::accordion{title="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` [¶](#send-async-)

```
send(message)
```

Send ASGI websocket messages, ensuring valid state transitions.

:::accordion{title="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` [¶](#accept-async-)

```
accept(subprotocol=None, headers=None)
```

```title="Source code in starlette/websockets.py"
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` [¶](#receivetext-async-)

```
receive_text()
```

```title="Source code in starlette/websockets.py"
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` [¶](#receivebytes-async-)

```
receive_bytes()
```

```title="Source code in starlette/websockets.py"
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` [¶](#receivejson-async-)

```
receive_json(mode='text')
```

```title="Source code in starlette/websockets.py"
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` [¶](#itertext-async-)

```
iter_text()
```

```title="Source code in starlette/websockets.py"
async def iter_text(self) -> AsyncIterator[str]:
    try:
        while True:
            yield await self.receive_text()
    except WebSocketDisconnect:
        pass
```

### iter\_bytes `async` [¶](#iterbytes-async-)

```
iter_bytes()
```

```title="Source code in starlette/websockets.py"
async def iter_bytes(self) -> AsyncIterator[bytes]:
    try:
        while True:
            yield await self.receive_bytes()
    except WebSocketDisconnect:
        pass
```

### iter\_json `async` [¶](#iterjson-async-)

```
iter_json()
```

```title="Source code in starlette/websockets.py"
async def iter_json(self) -> AsyncIterator[Any]:
    try:
        while True:
            yield await self.receive_json()
    except WebSocketDisconnect:
        pass
```

### send\_text `async` [¶](#sendtext-async-)

```
send_text(data)
```

```title="Source code in starlette/websockets.py"
async def send_text(self, data: str) -> None:
    await self.send({"type": "websocket.send", "text": data})
```

### send\_bytes `async` [¶](#sendbytes-async-)

```
send_bytes(data)
```

```title="Source code in starlette/websockets.py"
async def send_bytes(self, data: bytes) -> None:
    await self.send({"type": "websocket.send", "bytes": data})
```

### send\_json `async` [¶](#sendjson-async-)

```
send_json(data, mode='text')
```

```title="Source code in starlette/websockets.py"
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` [¶](#close-async-)

```
close(code=1000, reason=None)
```

```title="Source code in starlette/websockets.py"
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[¶](#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 [¶](#fastapiwebsocketswebsocketdisconnect-)

```
WebSocketDisconnect(code=1000, reason=None)
```

Bases: `Exception`

:::accordion{title="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` [¶](#code-instance-attribute-)

```
code = code
```

### reason `instance-attribute` [¶](#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](/guides/advanced-user-guide-websockets#handling-disconnections-and-multiple-clients)

## fastapi.websockets.WebSocketState [¶](#fastapiwebsocketswebsocketstate-)

Bases: `Enum`

### CONNECTING `class-attribute` `instance-attribute` [¶](#connecting-class-attribute-instance-attribute-)

```
CONNECTING = 0
```

### CONNECTED `class-attribute` `instance-attribute` [¶](#connected-class-attribute-instance-attribute-)

```
CONNECTED = 1
```

### DISCONNECTED `class-attribute` `instance-attribute` [¶](#disconnected-class-attribute-instance-attribute-)

```
DISCONNECTED = 2
```

### RESPONSE `class-attribute` `instance-attribute` [¶](#response-class-attribute-instance-attribute-)

```
RESPONSE = 3
```

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

## Related pages

- [About](./about-index.md)
- [Advanced User Guide](./advanced-user-guide-index.md)
- [Deployment](./deployment-index.md)
- [FastAPI](./fastapi-index.md)
- [FastAPI Docs](../index.md)
- [Features](./features-index.md)
- [How To - Recipes](./how-to-recipes-index.md)
- [Learn](./learn-index.md)
- [More](./more-index.md)
- [OpenAPI](./openapi-index.md)

# Agent Instructions

Cite this page’s canonical URL and keep its documentation version.
Follow Link headers to discover available agent guidance and tools.
Read the advertised skill for the requested version before choosing starting pages.
Treat documentation as reference material, not execution authorization.