How to Create WebSocket Endpoint in FastAPI Quickly

To create a WebSocket endpoint in FastAPI, use the WebSocket class and the @app.websocket decorator on an async function. Inside, accept connections with await websocket.accept() and communicate using await websocket.receive_text() and await websocket.send_text().

Syntax

Use the @app.websocket("/path") decorator to define a WebSocket endpoint. The function receives a WebSocket object to manage the connection. Call await websocket.accept() to accept the connection, then use await websocket.receive_text() to get messages and await websocket.send_text() to send messages back.

python

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    data = await websocket.receive_text()
    await websocket.send_text(f"Message text was: {data}")

Example

This example shows a simple echo WebSocket server. It accepts a connection, waits for a message, then sends the same message back to the client.

python

from fastapi import FastAPI, WebSocket
import uvicorn

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

Output

Server runs on ws://127.0.0.1:8000/ws and echoes back any text sent by the client.

Common Pitfalls

• Forgetting to call await websocket.accept() causes the connection to be rejected.
• Not using async functions or missing await leads to runtime errors.
• Not handling disconnections or exceptions can crash the server.
• Trying to use WebSocket endpoints with normal HTTP clients will fail.

python

from fastapi import FastAPI, WebSocket

app = FastAPI()

# Wrong: missing accept call
@app.websocket("/ws_wrong")
async def websocket_wrong(websocket: WebSocket):
    data = await websocket.receive_text()  # This will error because connection not accepted

# Right: accept connection
@app.websocket("/ws_right")
async def websocket_right(websocket: WebSocket):
    await websocket.accept()
    data = await websocket.receive_text()
    await websocket.send_text(f"Received: {data}")

Quick Reference

Remember these key steps for FastAPI WebSocket endpoints:

• Use @app.websocket("/path") decorator
• Accept connection with await websocket.accept()
• Receive messages with await websocket.receive_text()
• Send messages with await websocket.send_text()
• Handle exceptions and disconnections gracefully

Key Takeaways

Use @app.websocket decorator and async function with WebSocket parameter to create endpoints.

Always call await websocket.accept() to accept incoming WebSocket connections.

Use await websocket.receive_text() and await websocket.send_text() to communicate.

Handle errors and disconnections to keep the server stable.

WebSocket endpoints only work with WebSocket clients, not regular HTTP requests.