Socket.IO Server (Python)

~12 min · library, python-socketio, fastapi

Level 0Poller

0 XP0/60 lessons0/10 achievements

0/120 XP to next level120 XP to go0% complete

python-socketio + FastAPI

The standard Python Socket.IO server is python-socketio. It supports asgi, integrates with FastAPI, and ships a Redis manager for horizontal scaling. The pattern: build the FastAPI app for your REST endpoints, build a Socket.IO server with AsyncServer, wrap both in ASGIApp, and uvicorn serves the lot.

Auth in connect

Socket.IO's connect handler receives an auth dict separately from query string and headers — the client passes it via the SDK's auth option. Validate; if invalid, raise ConnectionRefusedError and the client never sees the connection succeed. Save user identity to sio.save_session(sid, ...) for use in later handlers.

Code

FastAPI + Socket.IO server·python

import socketio
from fastapi import FastAPI

# REST app
api = FastAPI()

@api.get('/health')
async def health():
    return {'ok': True}

# Socket.IO server
sio = socketio.AsyncServer(async_mode='asgi', cors_allowed_origins='*')

@sio.event
async def connect(sid, environ, auth):
    token = (auth or {}).get('token')
    if not token or not validate_token(token):
        raise socketio.exceptions.ConnectionRefusedError('unauthorized')
    user = decode(token)
    await sio.save_session(sid, {'user': user})
    print('connected', sid, user['name'])

@sio.event
async def disconnect(sid):
    print('disconnected', sid)

@sio.on('chat:message')
async def chat_message(sid, data):
    sess = await sio.get_session(sid)
    user = sess['user']
    await sio.emit('chat:message', {
        'from': user['name'],
        'text': data['text'],
    }, room=data['room'])

# Mount everything together
app = socketio.ASGIApp(sio, other_asgi_app=api)
# Run: uvicorn main:app --host 0.0.0.0 --port 8000

External links

python-socketio docs

Exercise

Build the server above. Connect from a Socket.IO client with a valid token, then with an invalid token. Confirm valid connects succeed and the bad one shows a connect_error in the client. Without the SDK on the client side, you cannot connect at all — confirm that too.