Skip to main content

​
Overview

FastrAPI provides built-in support for WebSocket connections, allowing real-time bidirectional communication between clients and servers. The WebSocket API is designed to be simple and intuitive, with support for text, binary, and JSON messages.

​
Decorator

​
@app.websocket()

Defines a WebSocket endpoint at the specified path.
​
path
string
required
The URL path for the WebSocket endpoint (e.g., “/ws”)
from fastrapi import FastrAPI

app = FastrAPI()

@app.websocket("/ws")
async def websocket_endpoint(ws):
    await ws.accept()
    # Handle WebSocket connection

​
WebSocket class

The WebSocket object is passed to your handler function and provides methods for managing the connection.

​
accept()

Accepts the WebSocket connection. Must be called before sending or receiving messages.
​
return
None
Returns nothing. The connection is established.
@app.websocket("/ws")
async def handler(ws):
    await ws.accept()  # Accept the connection
    # Now you can send/receive messages

​
send_text()

Sends a text message to the client.
​
data
string
required
The text message to send
​
return
Coroutine
Returns an awaitable that completes when the message is sent
await ws.send_text("Hello, client!")

​
send_bytes()

Sends binary data to the client.
​
data
bytes
required
The binary data to send
​
return
Coroutine
Returns an awaitable that completes when the data is sent
binary_data = b"\x00\x01\x02\x03"
await ws.send_bytes(binary_data)

​
send_json()

Serializes a Python object to JSON and sends it as a text message.
​
data
Any
required
Any JSON-serializable Python object (dict, list, etc.)
​
return
Coroutine
Returns an awaitable that completes when the message is sent
await ws.send_json({"type": "notification", "message": "Hello!"})

​
receive_text()

Receives a text message from the client.
​
return
string
The text message received from the client
Raises:
  • ConnectionError: If the WebSocket is closed
  • TypeError: If the received message is binary instead of text
message = await ws.receive_text()
print(f"Client said: {message}")

​
receive_bytes()

Receives binary data from the client.
​
return
bytes
The binary data received from the client
Raises:
  • ConnectionError: If the WebSocket is closed
  • TypeError: If the received message is text instead of binary
data = await ws.receive_bytes()
print(f"Received {len(data)} bytes")

​
receive_json()

Receives a text message and parses it as JSON.
​
return
Any
The parsed JSON object (dict, list, etc.)
Raises:
  • ConnectionError: If the WebSocket is closed
  • TypeError: If the received message is binary instead of text
  • ValueError: If the message is not valid JSON
data = await ws.receive_json()
print(f"Received: {data['message']}")

​
close()

Closes the WebSocket connection.
​
code
int
WebSocket close code (optional)
​
return
Coroutine
Returns an awaitable that completes when the connection is closed
await ws.close()
# Or with a specific close code
await ws.close(1000)  # Normal closure

​
Properties

​
client_state

Returns the current state of the WebSocket connection.
​
client_state
int
Connection state: 1 (connected) or 3 (disconnected)
if ws.client_state == 1:
    print("Connected")
else:
    print("Disconnected")

​
Complete example

from fastrapi import FastrAPI

app = FastrAPI()

@app.websocket("/ws")
async def websocket_endpoint(ws):
    await ws.accept()
    
    try:
        while ws.client_state == 1:
            # Receive a message
            data = await ws.receive_text()
            print(f"Client said: {data}")
            
            # Send a JSON response
            await ws.send_json({
                "reply": "Message received",
                "echo": data
            })
            
    except Exception as e:
        print(f"Connection closed: {e}")
    finally:
        await ws.close()

if __name__ == "__main__":
    app.serve(host="0.0.0.0", port=8000)

​
Error handling

WebSocket connections can close unexpectedly. Always wrap your WebSocket logic in try-except blocks: 
@app.websocket("/ws")
async def handler(ws):
    await ws.accept()
    
    try:
        while True:
            message = await ws.receive_text()
            await ws.send_text(f"Echo: {message}")
    except ConnectionError:
        print("Client disconnected")
    except Exception as e:
        print(f"Error: {e}")
        await ws.close()

Build docs developers (and LLMs) love

Get started for freeTalk to us