# ws_bridge

URL: /pounce/api/asgi/ws_bridge/
Section: asgi
Description: WebSocket ASGI bridge — translates between WSProtocol events and ASGI.

Builds ASGI ``websocket`` scope dicts and creates the async receive/send
callables for WebSocket ASGI apps.

ASGI WebSocket lifecycle:
    1. App receives ``websocket.connect``
    2. App sends ``websocket.accept`` (or ``websocket.close`` to reject)
    3. App receives ``websocket.receive`` messages
    4. App sends ``websocket.send`` messages
    5. Either side sends ``websocket.close`` / receives ``websocket.disconnect``

---

> For a complete page index, fetch /pounce/llms.txt.

Open LLM text
(/pounce/api/asgi/ws_bridge/index.txt)

Share with AI

Ask Claude
(https://claude.ai/new?q=Please%20help%20me%20understand%20this%20documentation%3A%20%2Fpounce%2Fapi%2Fasgi%2Fws_bridge%2Findex.txt)

Ask ChatGPT
(https://chatgpt.com/?q=Please%20help%20me%20understand%20this%20documentation%3A%20%2Fpounce%2Fapi%2Fasgi%2Fws_bridge%2Findex.txt)

Ask Gemini
(https://gemini.google.com/app?q=Please%20help%20me%20understand%20this%20documentation%3A%20%2Fpounce%2Fapi%2Fasgi%2Fws_bridge%2Findex.txt)

Ask Copilot
(https://copilot.microsoft.com/?q=Please%20help%20me%20understand%20this%20documentation%3A%20%2Fpounce%2Fapi%2Fasgi%2Fws_bridge%2Findex.txt)

Module

#
`asgi.ws_bridge`

WebSocket ASGI bridge — translates between WSProtocol events and ASGI.

Builds ASGI`websocket`scope dicts and creates the async receive/send
callables for WebSocket ASGI apps.

ASGI WebSocket lifecycle:

```
1. App receives ``websocket.connect``
2. App sends ``websocket.accept`` (or ``websocket.close`` to reject)
3. App receives ``websocket.receive`` messages
4. App sends ``websocket.send`` messages
5. Either side sends ``websocket.close`` / receives ``websocket.disconnect``
```

3Functions

## Functions

`build_ws_scope`

5

`dict[str, Any]`

▼

Build an ASGI WebSocket scope dict from the upgrade request.

`def build_ws_scope(request: RequestReceived, config: ServerConfig, client: tuple[str, int], server: tuple[str, int], *, state: dict[str, Any] | None = None) -> dict[str, Any]`

##### Parameters

Name
Type
Description

`request`
`RequestReceived (/pounce/api/protocols/_base/#RequestReceived)`

The parsed HTTP upgrade request.

`config`
`ServerConfig (/pounce/api/config/#ServerConfig)`

Server configuration.

`client`
`tuple[str, int]`

Client (host, port) tuple.

`server`
`tuple[str, int]`

Server (host, port) tuple.

`state`
`dict[str, Any] | None`

Default:`None`

##### Returns

`dict[str, Any]`

`create_ws_receive`

1

`Receive`

▼

Create an ASGI receive callable for WebSocket.

The worker pushes WebSocket eve…

`def create_ws_receive(events: asyncio.Queue[dict[str, Any]]) -> Receive`

Create an ASGI receive callable for WebSocket.

The worker pushes WebSocket events into the queue. The ASGI app
consumes them via`receive()`.

##### Parameters

Name
Type
Description

`events`
`asyncio.Queue[dict[str, Any]]`

Queue of ASGI WebSocket event dicts.

##### Returns

`Receive`

`create_ws_send`

6

`Send`

▼

Create an ASGI send callable for WebSocket.

Handles ``websocket.accept``, ``we…

`def create_ws_send(writer: asyncio.StreamWriter, ws_protocol: WSProtocol, ws_key: bytes, *, accept_event: asyncio.Event, close_event: asyncio.Event, write_timeout: float = 30.0) -> Send`

Create an ASGI send callable for WebSocket.

Handles`websocket.accept`, `websocket.send`, and
`websocket.close` (/pounce/api/protocols/ws/#WSProtocol) messages from the ASGI app.

##### Parameters

Name
Type
Description

`writer`
`asyncio.StreamWriter`

Asyncio stream writer for the connection.

`ws_protocol`
`WSProtocol (/pounce/api/protocols/ws/#WSProtocol)`

The WSProtocol instance for WebSocket framing.

`ws_key`
`bytes`

The Sec-WebSocket-Key from the client's upgrade request.

`accept_event`
`asyncio.Event`

Set when the app sends`websocket.accept`.

`close_event`
`asyncio.Event`

Set when the app sends`websocket.close`.

`write_timeout`
`float`

Default:`30.0`

##### Returns

`Send`