python-socketio
python-socketio is a Python implementation of the Socket.IO realtime protocol for both servers and clients. It provides synchronous and asyncio APIs, event handlers, acknowledgements, rooms, namespaces, per-connection sessions, reconnection, long-polling and WebSocket transports, WSGI and ASGI wrappers, and message-queue managers for multi-process delivery. Socket.IO is its own event protocol layered over Engine.IO; it is not a generic WebSocket library.
The right Python endpoint when Socket.IO is a product requirement and its richer event model earns the operational cost. For plain WebSocket, SSE, or durable messaging, choose the protocol built for that job instead of treating Socket.IO as all realtime infrastructure.
Use it if
- Your browser, mobile, Node, or Python clients already speak Socket.IO protocol revision 5 and need a Python peer
- You want named events, acknowledgements, automatic reconnects, rooms, namespaces, and polling fallback rather than raw WebSocket frames
- You need a synchronous WSGI deployment or an asyncio ASGI, aiohttp, Tornado, or other documented server integration
- You are prepared to add Redis, RabbitMQ, or another manager when multiple processes or background workers must emit to the same clients
- Your peer speaks plain WebSocket: Socket.IO adds Engine.IO framing and event semantics, so a normal WebSocket client cannot connect directly
- You only need one-way server updates: Server-Sent Events are simpler through proxies and do not require the polling, upgrade, room, and acknowledgement machinery
- You need durable messages, replay, offline delivery, or ordered job processing: rooms and emits are live connection features, not a persistent broker
- You expect horizontal scaling to work by adding workers: the deployment docs require compatible routing or WebSocket-only clients plus a message queue so processes can coordinate
- You plan to start a new eventlet deployment: the official docs say eventlet is in maintenance mode and explicitly do not recommend it for new projects
Setup reality
`pip install python-socketio` installs the core server with python-engineio and bidict, but your chosen deployment adds more packages and decisions. Version 5.16.4 requires Python 3.8 or newer. The synchronous client needs the `client` extra for requests and websocket-client; the asyncio client needs `asyncio-client` for aiohttp. Servers still need an actual WSGI or ASGI host, and integrations such as aiohttp, Tornado, gevent, Redis, or Kombu are installed separately. Match protocol generations before debugging anything else: the README says python-socketio 5.x matches JavaScript Socket.IO 3.x and 4.x, Socket.IO protocol 5, and Engine.IO protocol 4. Same-package Python clients and servers are the easiest pairing. Choose `Server` with `WSGIApp` for threaded WSGI or `AsyncServer(async_mode='asgi')` with `ASGIApp` for asyncio; event handlers and emits must follow the matching sync or awaitable API. Authenticate in `connect` using the request environment and client `auth` payload, and reject unauthorized clients before accepting events. Same-origin checking is on by default. Use an explicit origin allowlist for cross-origin frontends; the docs warn that `cors_allowed_origins='*'` can introduce CSRF exposure. Reverse proxies must pass WebSocket upgrades and use timeouts long enough for persistent connections. Per-client sessions disappear on disconnect and are not restored after reconnect, so durable user state belongs elsewhere. Multi-worker Gunicorn adds constraints: long-polling is incompatible with its routing unless clients use WebSocket directly, and every process needs a shared message-queue manager. Redis, RabbitMQ, Kafka, or other manager choices bring credentials and operations. Finally, emits are ephemeral unless your application stores and retries them; acknowledgements indicate client handling, not durable business completion.
Patterns
Create an asyncio ASGI applicationcreate-asgi-server
import socketio
sio = socketio.AsyncServer(
async_mode='asgi',
cors_allowed_origins=['https://app.example.com'],
)
app = socketio.ASGIApp(sio)Run app with an ASGI server such as Uvicorn. Use an explicit origin list instead of '*' for browser clients.
Authenticate before accepting a clientauthenticate-connection
from socketio.exceptions import ConnectionRefusedError
@sio.event
async def connect(sid, environ, auth):
token = (auth or {}).get('token')
user = await authenticate(token)
if user is None:
raise ConnectionRefusedError('authentication failed')
await sio.save_session(sid, {'user_id': user.id})The auth payload is client-controlled. Verify it server-side, and do not rely on a Socket.IO session surviving reconnection.
Handle an event and acknowledge ithandle-event
@sio.event
async def update_profile(sid, data):
session = await sio.get_session(sid)
await save_profile(session['user_id'], data)
return {'ok': True}Returning values sends an acknowledgement only when the client requested one. Validate event payloads because the library does not impose a schema.
Send an event to one clientemit-to-client
await sio.emit(
'report_ready',
{'report_id': report_id},
to=user_sid,
)Omitting to broadcasts to every connected client. Treat session IDs as connection identifiers, not durable user IDs.
Wait for a client acknowledgementawait-acknowledgement
reply = await sio.call(
'confirm_action',
{'action_id': action_id},
to=user_sid,
timeout=10,
)call is for a single client and can time out. An acknowledgement proves handler execution, not that downstream business work was durably committed.
Join and broadcast to a roombroadcast-room
sync_sio = socketio.Server(async_mode='threading')
@sync_sio.event
def join_project(sid, project_id):
authorize_project(sid, project_id)
sync_sio.enter_room(sid, f'project:{project_id}')
@sync_sio.event
def project_message(sid, data):
sync_sio.emit('project_message', data, room=f"project:{data['project_id']}", skip_sid=sid)Room names are application data, not authorization. Check membership rights before entering or emitting to a room.
Update per-connection session stateuse-connection-session
@sio.event
async def select_workspace(sid, workspace_id):
async with sio.session(sid) as session:
session['workspace_id'] = workspace_idSessions are scoped to a namespace and destroyed on disconnect. Store durable preferences or login state in an external database.
Coordinate ASGI workers through Redisscale-with-redis
import os
import socketio
manager = socketio.AsyncRedisManager(os.environ['REDIS_URL'])
sio = socketio.AsyncServer(
async_mode='asgi',
client_manager=manager,
transports=['websocket'],
)
app = socketio.ASGIApp(sio)Install redis separately. Multiple workers also need compatible proxy routing, and forcing WebSocket removes the long-polling fallback.
Emit from a background worker processemit-from-worker
import os
import socketio
external = socketio.RedisManager(
os.environ['REDIS_URL'],
write_only=True,
)
external.emit('job_finished', {'job_id': job_id}, room=user_room)The worker and servers must use the same queue and channel configuration. This emit is still not a durable event log.
Connect a synchronous Python clientconnect-sync-client
import socketio
client = socketio.Client(reconnection=True)
client.connect(
'https://realtime.example.com',
auth={'token': token},
transports=['websocket'],
)
client.emit('subscribe', {'project_id': '42'})
client.wait()Install python-socketio[client]. Forcing WebSocket requires proxy support and gives up automatic polling fallback.
Receive events with AsyncClientconnect-async-client
import socketio
client = socketio.AsyncClient()
@client.on('report_ready')
async def report_ready(data):
await download(data['report_id'])
await client.connect('https://realtime.example.com', auth={'token': token})
await client.wait()Install python-socketio[asyncio-client]. Event handlers run in the client's asyncio environment and should not call blocking I/O.
Enable Socket.IO and Engine.IO logsdebug-protocol
import logging
import socketio
logger = logging.getLogger('realtime')
sio = socketio.AsyncServer(
async_mode='asgi',
logger=logger,
engineio_logger=logger,
)Protocol logs can include connection details and payload context. Use a controlled logger and avoid verbose production logging of secrets.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| websockets | PyPI | Both ends can use plain WebSocket and you want a focused asyncio implementation without Socket.IO framing |
| aiohttp | PyPI | Your asyncio service already uses aiohttp and needs its native HTTP and WebSocket support rather than Socket.IO clients |
| flask-socketio | PyPI | You are building specifically on Flask and want framework-aware handlers and deployment guidance around this engine |