Fix WebSocketDisconnect: starlette.websockets.WebSocketDisconnect: 1000 in FastAPI

This error occurs when a WebSocket client disconnects and the server tries to send or receive data on the closed connection without handling the disconnection. Fix it by wrapping your WebSocket receive/send loop in a try/except block that catches WebSocketDisconnect and performs cleanup instead of crashing.

Reading the Stack Trace

Here's what each line means:

• File "/app/src/routes/chat.py", line 22, in websocket_chat: The WebSocket endpoint is blocked waiting for receive_text() when the client disconnects, raising WebSocketDisconnect.
• File "/app/venv/lib/python3.11/site-packages/starlette/websockets.py", line 131, in receive_text: Starlette's receive_text raises WebSocketDisconnect when the underlying connection is closed by the client.
• starlette.websockets.WebSocketDisconnect: 1000: Code 1000 indicates a normal closure by the client. The server should handle this gracefully rather than letting it propagate as an unhandled exception.

Common Causes

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

class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)  # Fails on disconnected clients

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        await websocket.send_text("Goodbye")  # Cannot send to closed socket

The Fix

Wrap the receive loop in a try/except that catches WebSocketDisconnect. On disconnect, remove the client from the connection manager and notify remaining clients. This prevents the unhandled exception and properly cleans up resources.

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

from starlette.websockets import WebSocketDisconnect

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await websocket.accept()
    manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Message: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast("A user left the chat.")

Testing the Fix

import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)


def test_websocket_connect_and_send():
    with client.websocket_connect("/ws/chat") as websocket:
        websocket.send_text("Hello")
        data = websocket.receive_text()
        assert "Hello" in data


def test_websocket_disconnect_handled_gracefully():
    with client.websocket_connect("/ws/chat") as websocket:
        websocket.send_text("Hello")
        websocket.receive_text()
    # No exception should be raised after the context manager closes


def test_websocket_multiple_messages():
    with client.websocket_connect("/ws/chat") as websocket:
        for i in range(5):
            websocket.send_text(f"Message {i}")
            data = websocket.receive_text()
            assert f"Message {i}" in data

Run your tests:

pytest tests/test_websocket.py -v

Pushing Through CI/CD

git checkout -b fix/fastapi-websocket-disconnect,git add src/routes/chat.py tests/test_websocket.py,git commit -m "fix: handle WebSocketDisconnect with try/except and cleanup",git push origin fix/fastapi-websocket-disconnect

Your CI config should look something like this:

name: CI
on:
  pull_request:
    branches: [main]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
          cache: 'pip'
      - run: pip install -r requirements.txt
      - run: pytest --tb=short -q

The Full Manual Process: 18 Steps

Here's every step you just went through to fix this one bug:

1. Notice the error alert or see it in your monitoring tool
2. Open the error dashboard and read the stack trace
3. Identify the file and line number from the stack trace
4. Open your IDE and navigate to the file
5. Read the surrounding code to understand context
6. Reproduce the error locally
7. Identify the root cause
8. Write the fix
9. Run the test suite locally
10. Fix any failing tests
11. Write new tests covering the edge case
12. Run the full test suite again
13. Create a new git branch
14. Commit and push your changes
15. Open a pull request
16. Wait for code review
17. Merge and deploy to production
18. Monitor production to confirm the error is resolved

Total time: 30-60 minutes. For one bug.

pip install bugstack

import bugstack

bugstack.init(api_key=os.environ["BUGSTACK_API_KEY"])

Deploying the Fix (Manual Path)

1. Run the test suite locally to confirm WebSocket connections and disconnections are handled.
2. Open a pull request with the disconnect handling fix.
3. Wait for CI checks to pass on the PR.
4. Have a teammate review and approve the PR.
5. Merge to main and verify WebSocket stability under load in staging.

Frequently Asked Questions