PYTHON-ENGINEIO(1) python-engineio PYTHON-ENGINEIO(1) (NAME) python-engineio - python-engineio Engine.IO . (GETTING STARTED) Engine.IO (What is Engine.IO?) Engine.IO ( ) . JavaScript . asyncio . Engine.IO . "" . (Event handlers) . (Client Examples) : import engineio eio = engineio.Client() @eio.on('connect') def on_connect(): print('connection established') @eio.on('message') def on_message(data): print('message received with ', data) eio.send({'response': 'my response'}) @eio.on('disconnect') def on_disconnect(): print('disconnected from server') eio.connect('http://localhost:5000') eio.wait() Engine.IO : o Engine.IO . o Python 3.6+. o Python asyncio. o . o HTTP long-polling WebSocket. Eventlet : import engineio import eventlet eio = engineio.Server() app = engineio.WSGIApp(eio, static_files={ '/': {'content_type': 'text/html', 'filename': 'index.html'} }) @eio.on('connect') def connect(sid, environ): print("connect ", sid) @eio.on('message') def message(sid, data): print("message ", data) eio.send(sid, 'reply') @eio.on('disconnect') def disconnect(sid): print('disconnect ', sid) if __name__ == '__main__': eventlet.wsgi.server(eventlet.listen(('', 5000)), app) asyncio Uvicorn : import engineio import uvicorn eio = engineio.AsyncServer() app = engineio.ASGIApp(eio, static_files={ '/': {'content_type': 'text/html', 'filename': 'index.html'} }) @eio.on('connect') def connect(sid, environ): print("connect ", sid) @eio.on('message') async def message(sid, data): print("message ", data) await eio.send(sid, 'reply') @eio.on('disconnect') def disconnect(sid): print('disconnect ', sid) if __name__ == '__main__': uvicorn.run('127.0.0.1', 5000) o Engine.IO . o Python 3.6+. o asyncio. o (asynchronous) . o WSGI ASGI Gunicorn Uvicorn eventlet gevent . o WSGI Flask Django . o asyncio aiohttp sanic tornado . o (event-based) . o HTTP long-polling WebSocket. o XHR2 XHR . o (). o HTTP gzip deflate. o CORS (cross-origin) . ENGINE.IO (THE ENGINE.IO CLIENT) Engine.IO : o engineio.Client() <#engineio.Client> . o engineio.AsyncClient() <#engineio.AsyncClient> asyncio . asyncio (coroutine) . (Installation) : pip install "python-engineio[client]" asyncio : pip install "python-engineio[asyncio_client]" (Creating a Client Instance) Engine.IO : import engineio # standard Python eio = engineio.Client() # asyncio eio = engineio.AsyncClient() on : @eio.on('connect') def on_connect(): print('I'm connected!') @eio.on('message') def on_message(data): print('I received a message!') @eio.on('disconnect') def on_disconnect(reason): print('I'm disconnected! reason:', reason) asyncio : @eio.on('message') async def on_message(data): print('I received a message!') on . connect message disconnect. data 'message' . disconnect ( ) . . : @eio.on('disconnect') def on_disconnect(reason): if reason == eio.reason.CLIENT_DISCONNECT: print('client disconnection') elif reason == eio.reason.SERVER_DISCONNECT: print('the server kicked me out') else: print(f'disconnect reason: {reason}') (Connecting to a Server) connect() : eio.connect('http://localhost:5000') asyncio : await eio.connect('http://localhost:5000') . sid : print('my sid is', eio.sid) send() : eio.send({'foo': 'bar'}) asyncio (coroutine): await eio.send({'foo': 'bar'}) . str bytes dict list . . send() (event handler) . (Disconnecting from the Server) disconnect() : eio.disconnect() asyncio : await eio.disconnect() (Managing Background Tasks) (background task) . (main thread) Engine.IO . wait() : eio.wait() asyncio: await eio.wait() : def my_background_task(my_argument) # do some background work here! pass eio.start_background_task(my_background_task, 123) (positional) (keyword) . asyncio : async def my_background_task(my_argument) # do some background work here! pass eio.start_background_task(my_background_task, 123) . sleep() : eio.sleep(2) asyncio: await eio.sleep(2) (sleep) ..SS " (Debugging and Troubleshooting)" : import engineio # standard Python eio = engineio.Client(logger=True) # asyncio eio = engineio.AsyncClient(logger=True) logger : o True: stderr INFO . o False: stderr ERROR . . o logging.Logger: . . ENGINE.IO (THE ENGINE.IO SERVER) Engine.IO : o engineio.Server() <#engineio.Server> . o engineio.AsyncServer() <#engineio.AsyncServer> asyncio . asyncio (coroutine) . (Installation) Engine.IO : pip install "python-engineio" (asynchronous) . (Deployment Strategies) . (Creating a Server Instance) Engine.IO engineio.Server <#engineio.Server> . engineio.WSGIApp <#engineio.WSGIApp> WSGI : import engineio # create a Engine.IO server eio = engineio.Server() # wrap with a WSGI application app = engineio.WSGIApp(eio) asyncio engineio.AsyncServer <#engineio.AsyncServer> . engineio.ASGIApp <#engineio.ASGIApp> ASGI : # create a Engine.IO server eio = engineio.AsyncServer() # wrap with ASGI application app = engineio.ASGIApp(eio) (middleware) Engine.IO . Engine.IO WSGI ASGI : from wsgi import app # a Flask, Django, etc. application app = engineio.WSGIApp(eio, app) Engine.IO . HTML CSS JavaScript . Python / . URL : static_files = { '/': 'latency.html', '/static/engine.io.js': 'static/engine.io.js', '/static/style.css': 'static/style.css', } / ( URL ) latency.html (content type) text/html . .html .css .js .json .jpg .png .gif .txt . application/octet-stream . : static_files = { '/': {'filename': 'latency.html', 'content_type': 'text/plain'}, } : static_files = { '/static': './public', } URL /static public /static/index.html ./public/index.html /static/css/styles.css ./public/css/styles.css . URL / index.html . /static/ ./public/index.html . : static_files = { '/static': './public', '': 'image.gif', } /static/ ./public/image.gif . : static_files = { '/static': './public', '': {'filename': 'image.gif', 'content_type': 'text/plain'}, } static_files engineio.WSGIApp engineio.ASGIApp : # for standard WSGI applications eio = engineio.Server() app = engineio.WSGIApp(eio, static_files=static_files) # for asyncio-based ASGI applications eio = engineio.AsyncServer() app = engineio.ASGIApp(eio, static_files=static_files) : o Engine.IO . o . o Engine.IO 404 . : (caching) . (production) ..SS (Defining Event Handlers) on : @eio.on('connect') def on_connect(sid): print('A client connected!') @eio.on('message') def on_message(sid, data): print('I received a message!') @eio.on('disconnect') def on_disconnect(sid, reason): print('Client disconnected! reason:', reason) asyncio (coroutine) : @eio.on('message') async def on_message(sid, data): print('I received a message!') on . connect message disconnect. sid . sid . connect . . ( None ) True . False JSON ( ) . 401 . data 'message' . disconnect . . : @eio.on('disconnect') def on_disconnect(sid, reason): if reason == eio.reason.CLIENT_DISCONNECT: print('the client went away') elif reason == eio.reason.SERVER_DISCONNECT: print('the client was kicked out') else: print(f'disconnect reason: {reason}') (Sending Messages) send() : eio.send(sid, {'foo': 'bar'}) asyncio : await eio.send(sid, {'foo': 'bar'}) . . str bytes dict list . . send() ..SS (User Sessions) . . save_session() get_session() : @eio.on('connect') def on_connect(sid, environ): username = authenticate_user(environ) eio.save_session(sid, {'username': username}) @eio.on('message') def on_message(sid, data): session = eio.get_session(sid) print('message from ', session['username']) asyncio (coroutine) : @eio.on('connect') async def on_connect(sid, environ): username = authenticate_user(environ) await eio.save_session(sid, {'username': username}) @eio.on('message') async def on_message(sid, data): session = await eio.get_session(sid) print('message from ', session['username']) session() : @eio.on('connect') def on_connect(sid, environ): username = authenticate_user(environ) with eio.session(sid) as session: session['username'] = username @eio.on('message') def on_message(sid, data): with eio.session(sid) as session: print('message from ', session['username']) asyncio : @eio.on('connect') def on_connect(sid, environ): username = authenticate_user(environ) async with eio.session(sid) as session: session['username'] = username @eio.on('message') def on_message(sid, data): async with eio.session(sid) as session: print('message from ', session['username']) : ..SS (Disconnecting a Client) disconnect() sid : eio.disconnect(sid) asyncio : await eio.disconnect(sid) (Managing Background Tasks) : def my_background_task(my_argument) # do some background work here! pass eio.start_background_task(my_background_task, 123) (positional) (keyword) . asyncio : async def my_background_task(my_argument) # do some background work here! pass eio.start_background_task(my_background_task, 123) . sleep() : eio.sleep(2) asyncio: await eio.sleep(2) (sleep) ..SS : import engineio # standard Python eio = engineio.Server(logger=True) # asyncio eio = engineio.AsyncServer(logger=True) logger : o True: stderr INFO . o False: stderr ERROR . . o logging.Logger: . . Engine.IO . Uvicorn Daphne ASGI engineio.ASGIApp ASGI Engine.IO engineio.AsyncServer : eio = engineio.AsyncServer(async_mode='asgi') app = engineio.ASGIApp(eio) engineio.ASGIApp Engine.IO ASGI ASGI Engine.IO : eio = engineio.AsyncServer(async_mode='asgi') app = engineio.ASGIApp(eio, other_app) ASGIApp ASGI ASGI . Aiohttp aiohttp asyncio HTTP WebSocket . engineio.AsyncServer aiohttp . async_mode : eio = engineio.AsyncServer(async_mode='aiohttp') aiohttp : app = web.Application() eio.attach(app) aiohttp Engine.IO . () . aiohttp : if __name__ == '__main__': web.run_app(app) Tornado HTTP WebSocket . asyncio Tornado . engineio.AsyncServer tornado . async_mode : eio = engineio.AsyncServer(async_mode='tornado') tornado (request handler) Engine.IO : app = tornado.web.Application( [ (r"/engine.io/", engineio.get_tornado_handler(eio)), ], # ... other application options ) tornado Engine.IO . . tornado : app.listen(port) tornado.ioloop.IOLoop.current().start() Sanic : Sanic Sanic ASGI . Sanic Python . engineio.AsyncServer Sanic . async_mode : eio = engineio.AsyncServer(async_mode='sanic') Sanic : app = Sanic() eio.attach(app) Sanic Engine.IO . . Sanic : if __name__ == '__main__': app.run() CORS sanic-cors Sanic . CORS Sanic : eio = engineio.AsyncServer(async_mode='sanic', cors_allowed_origins=[]) Sanic CORS_SUPPORTS_CREDENTIALS : app.config['CORS_SUPPORTS_CREDENTIALS'] = True Eventlet Python 2 3 (coroutines) . Engine.IO eventlet long-polling WebSocket . engineio.Server eventlet . async_mode (constructor) : eio = engineio.Server(async_mode='eventlet') eventlet engineio.WSGIApp WSGI : app = engineio.WSGIApp(eio) import eventlet eventlet.wsgi.server(eventlet.listen(('', 8000)), app) Eventlet Gunicorn WSGI eventlet gunicorn . gunicorn : $ gunicorn -k eventlet -w 1 module:app gunicorn worker -w 1 . worker eventlet . gunicorn WebSocket WSGI . : Eventlet monkey_patch() . python-engineio monkey patching . Gevent Gevent eventlet . Engine.IO gevent long-polling websocket . engineio.Server eventlet gevent . gevent async_mode : eio = engineio.Server(async_mode='gevent') gevent engineio.WSGIApp WSGI : from gevent import pywsgi app = engineio.WSGIApp(eio) pywsgi.WSGIServer(('', 8000), app).serve_forever() WSGI gevent gunicorn (pure Python). gunicorn : $ gunicorn -k gevent -w 1 module:app eventlet (load balancing) gunicorn (worker process) -w 1 . gevent . : Gevent monkey_patch() (blocking) . python-engineio monkey patching . uWSGI uWSGI gevent Engine.IO WebSocket uWSGI . engineio.Server gevent uWSGI eventlet . async_mode (constructor) : # gevent with uWSGI eio = engineio.Server(async_mode='gevent_uwsgi') uWSGI . uWSGI . WebSocket WebSocket SSL . uWSGI latency.py : $ uwsgi --http :5000 --gevent 1000 --http-websockets --master --wsgi-file latency.py --callable app (Standard Threads) eventlet gevent Engine.IO . Werkzeug . engineio.Server eventlet gevent . async_mode : eio = engineio.Server(async_mode='threading') WSGI . Engine.IO Flask Flask Werkzeug : eio = engineio.Server(async_mode='threading') app = Flask(__name__) app.wsgi_app = engineio.WSGIApp(eio, app.wsgi_app) # ... Engine.IO and Flask handler functions ... if __name__ == '__main__': app.run() Engine.IO Gunicorn : $ gunicorn -w 1 --threads 100 module:app . WebSocket simple-websocket . WebSocket Werkzeug Gunicorn . WebSocket ..SS " (Scalability Notes)" Engine.IO (stateful) . Engine.IO : o Engine.IO . long-polling . worker . o (load balancer) . sticky sessions session affinity . (Cross-Origin Controls) (same-origin policy) . : o HTTP WebSocket Origin scheme host URL . 400 . o Origin . cors_allowed_origins . . '*' (CSRF) . (API REFERENCE) class engineio.Client(logger=False, json=None, request_timeout=5, http_session=None, ssl_verify=True, handle_sigint=True, websocket_extra_options=None, timestamp_requests=True) Engine.IO. Engine.IO websocket long-polling . o logger -- True logger . False . False . logger False . o json -- JSON . json dumps loads . JSON . o request_timeout -- ( ) . 5 . o http_session -- requests.Session . SSL CA . o ssl_verify -- True SSL False SSL . True . o handle_sigint -- True False . (main thread) . o websocket_extra_options -- websocket.create_connection() . o timestamp_requests -- True (timestamp) (cache-busting) query string Socket.IO . False . connect(url, headers=None, transports=None, engineio_path='engine.io') Engine.IO. o url -- (URL) Engine.IO. query string . o headers -- . o transports -- . 'polling' 'websocket' . polling websocket . o engineio_path -- (endpoint) Engine.IO . . : eio = engineio.Client() eio.connect('http://localhost:5000') wait() . . send(data) . data -- . str bytes list dict . list dict JSON . disconnect(abort=False, reason=None) . abort -- True . start_background_task(target, *args, **kwargs) . . o target -- . o args -- . o kwargs -- . join() . sleep(seconds=0) . create_queue(*args, **kwargs) (queue) . get_queue_empty_exception() create_queue() . create_event(*args, **kwargs) (event) . on(event, handler=None) (event handler) . (Parameters) o event -- . 'connect' 'message' 'disconnect' . o handler -- . . : # as a decorator: @eio.on('connect') def connect_handler(): print('Connection request') # as a method: def message_handler(msg): print('Received message: ', msg) eio.send('response') eio.on('message', message_handler) class reason . CLIENT_DISCONNECT = 'client disconnect' . SERVER_DISCONNECT = 'server disconnect' . TRANSPORT_ERROR = 'transport error' (transport). transport() . 'polling' 'websocket' . class engineio.AsyncClient(logger=False, json=None, request_timeout=5, http_session=None, ssl_verify=True, handle_sigint=True, websocket_extra_options=None, timestamp_requests=True) Engine.IO asyncio. Engine.IO websocket long-polling asyncio Python 3.5 . (Parameters) o logger -- True logger . False . False . False logger . o json -- JSON . json dumps loads . (process-wide) JSON . o request_timeout -- . 5 . o http_session -- aiohttp.ClientSession . SSL CA . o ssl_verify -- True SSL False SSL (self-signed). True . o handle_sigint -- True False . (main thread) . o websocket_extra_options -- aiohttp.ws_connect(). o timestamp_requests -- True (query string) Socket.IO (cache-busting) . False . async connect(url, headers=None, transports=None, engineio_path='engine.io') Engine.IO. (Parameters) o url -- (URL) Engine.IO. . o headers -- . o transports -- . 'polling' 'websocket'. polling websocket . o engineio_path -- (endpoint) Engine.IO . . : (coroutine) . : eio = engineio.Client() await eio.connect('http://localhost:5000') async wait() ..sp . : . async send(data) . data -- . str bytes list dict . list dict JSON . : . async disconnect(abort=False, reason=None) . abort -- True . : . start_background_task(target, *args, **kwargs) . . o target -- . o args -- . o kwargs -- . asyncio.Task . async sleep(seconds=0) . : . create_queue(*args, **kwargs) . get_queue_empty_exception() create_queue() . create_event() . on(event, handler=None) . o event -- . 'connect' 'message' 'disconnect' . o handler -- . . : # as a decorator: @eio.on('connect') def connect_handler(): print('Connection request') # as a method: def message_handler(msg): print('Received message: ', msg) eio.send('response') eio.on('message', message_handler) class reason . CLIENT_DISCONNECT = 'client disconnect' . SERVER_DISCONNECT = 'server disconnect' . TRANSPORT_ERROR = 'transport error' (transport). transport() . 'polling' 'websocket' . class engineio.Server(async_mode=None, ping_interval=25, ping_timeout=20, max_http_buffer_size=1000000, allow_upgrades=True, http_compression=True, compression_threshold=1024, cookie=None, cors_allowed_origins=None, cors_credentials=True, logger=False, json=None, async_handlers=True, monitor_clients=None, transports=None, **kwargs) Engine.IO. Engine.IO websocket long-polling . o async_mode -- . Deployment . "threading" "eventlet" "gevent" "gevent_uwsgi". "eventlet" "gevent_uwsgi" "gevent" "threading". . o ping_interval -- . 25 . (grace period) . o ping_timeout -- . 20 . o max_http_buffer_size -- . 1,000,000 . HTTP long-polling WebSocket ..IP o 2 allow_upgrades -- (transport) . True . o http_compression -- polling . True . o compression_threshold -- . 1024 . o cookie -- HTTP (session id) . 'name' . None () . o cors_allowed_origins -- (origin) . (same origin) . '*' ['*'] [] CORS . o cors_credentials -- ( ) . True . o logger -- True (logger) . False . False . logger False . o json -- JSON . JSON dumps loads . JSON . o async_handlers -- True . False . True . o monitor_clients -- True . False ( ). True . o transports -- . 'polling' 'websocket' . ['polling', 'websocket'] . o kwargs -- . send(sid, data) . (Parameters) o sid -- (session id) . o data -- . str bytes list dict . list dict JSON . send_packet(sid, pkt) (raw packet) . (Parameters) o sid -- (session id) . o pkt -- . get_session(sid) . (Parameters) sid -- (session id) . . save_session() (context manager) session . save_session(sid, session) . (Parameters) o sid -- (session id) . o session -- . session(sid) (context manager). (Parameters) sid -- (session id) . . . : @eio.on('connect') def on_connect(sid, environ): username = authenticate_user(environ) if not username: return False with eio.session(sid) as session: session['username'] = username @eio.on('message') def on_message(sid, msg): with eio.session(sid) as session: print('received message from ', session['username']) disconnect(sid=None) . (Parameters) sid -- (session id) . . handle_request(environ, start_response) HTTP . Engine.IO WSGI . Middleware <# engineio.Middleware> . (Parameters) o environ -- WSGI. o start_response -- start_response WSGI. HTTP . shutdown() Socket.IO. Socket.IO . . start_background_task(target, *args, **kwargs) . . o target -- . o args -- . o kwargs -- . join() . sleep(seconds=0) . . create_event(*args, **kwargs) . . create_queue(*args, **kwargs) . . generate_id() . get_queue_empty_exception() . . on(event, handler=None) . o event -- . 'connect' 'message' 'disconnect' . o handler -- . . : # as a decorator: @eio.on('connect') def connect_handler(sid, environ): print('Connection request') if environ['REMOTE_ADDR'] in blacklisted: return False # reject # as a method: def message_handler(sid, msg): print('Received message: ', msg) eio.send(sid, 'response') eio.on('message', message_handler) sid ( ) . 'connect' WSGI False . 'message' . 'disconnect' . class reason . CLIENT_DISCONNECT = 'client disconnect' . PING_TIMEOUT = 'ping timeout' . SERVER_DISCONNECT = 'server disconnect' . TRANSPORT_CLOSE = 'transport close' . TRANSPORT_ERROR = 'transport error' . transport(sid) . 'polling' 'websocket' . sid -- . class engineio.AsyncServer(async_mode=None, ping_interval=25, ping_timeout=20, max_http_buffer_size=1000000, allow_upgrades=True, http_compression=True, compression_threshold=1024, cookie=None, cors_allowed_origins=None, cors_credentials=True, logger=False, json=None, async_handlers=True, monitor_clients=None, transports=None, **kwargs) Engine.IO asyncio. Engine.IO websocket long-polling asyncio Python 3.5 . async_mode -- (asynchronous) . Deployment . "aiohttp" "sanic" "tornado" "asgi". "aiohttp" "sanic" "tornado" "asgi" . . o ping_interval -- . . (grace period) . o ping_timeout -- . . o max_http_buffer_size -- . . HTTP long-polling WebSocket . o allow_upgrades -- (transport upgrade). o http_compression -- polling. o compression_threshold -- . o cookie -- HTTP (session id) . 'name' (callable) (boolean) . None () . o cors_allowed_origins -- (origin) . (same origin) . '*' ['*'] CORS [] . o cors_credentials -- ( ) . o logger -- True (logger) . False . logger False . o json -- JSON . JSON dumps loads . (process-wide) JSON . o async_handlers -- True (threads) . False . True . o monitor_clients -- True . ( ) False . True . o transports -- . 'polling' 'websocket' . ['polling', 'websocket'] . o kwargs -- . attach(app, engineio_path='engine.io') Engine.IO (application). async send(sid, data) . (Parameters) o sid -- (session id) . o data -- . str bytes list dict . list dict JSON . : (coroutine) . async send_packet(sid, pkt) (raw packet) . (Parameters) o sid -- (session id) . o pkt -- . : (coroutine) . async get_session(sid) . (Parameters) sid -- (session id) . . . (context manager) session . async save_session(sid, session) . (Parameters) o sid -- (session id) . o session -- . session(sid) (syntax) (context manager). (Parameters) sid -- (session id) . (context manager) . . : @eio.on('connect') def on_connect(sid, environ): username = authenticate_user(environ) if not username: return False with eio.session(sid) as session: session['username'] = username @eio.on('message') def on_message(sid, msg): async with eio.session(sid) as session: print('received message from ', session['username']) async disconnect(sid=None) . sid -- (session id) . . : (coroutine) . async handle_request(*args, **kwargs) HTTP . Engine.IO . HTTP . : (coroutine) . async shutdown() Socket.IO. Socket.IO . . start_background_task(target, *args, **kwargs) (async) . async . o target -- . o args -- . o kwargs -- . asyncio.Task . async sleep(seconds=0) () (async) . async . : (coroutine) . create_queue(*args, **kwargs) (queue) (async) . async . async asyncio asyncio.Queue . get_queue_empty_exception() (async) . async . async asyncio asyncio.QueueEmpty . create_event(*args, **kwargs) (event) (async) . async . async asyncio asyncio.Event . generate_id() (session id) . on(event, handler=None) (event handler). o event -- . 'connect' 'message' 'disconnect' . o handler -- . . : # as a decorator: @eio.on('connect') def connect_handler(sid, environ): print('Connection request') if environ['REMOTE_ADDR'] in blacklisted: return False # reject # as a method: def message_handler(sid, msg): print('Received message: ', msg) eio.send(sid, 'response') eio.on('message', message_handler) sid ( ) . 'connect' WSGI False . 'message' . 'disconnect' . class reason . CLIENT_DISCONNECT = 'client disconnect' . PING_TIMEOUT = 'ping timeout' (Ping timeout). SERVER_DISCONNECT = 'server disconnect' . TRANSPORT_CLOSE = 'transport close' (Transport close). TRANSPORT_ERROR = 'transport error' (Transport error). transport(sid) (transport) ..sp 'polling' 'websocket' . sid -- . class engineio.WSGIApp(engineio_app, wsgi_app=None, static_files=None, engineio_path='engine.io') WSGI Engine.IO. Engine.IO . HTTP WSGI . o engineio_app -- Engine.IO. engineio.Server . o wsgi_app -- WSGI . o static_files -- . . o engineio_path -- (endpoint) Engine.IO . . : import engineio import eventlet eio = engineio.Server() app = engineio.WSGIApp(eio, static_files={ '/': {'content_type': 'text/html', 'filename': 'index.html'}, '/index.html': {'content_type': 'text/html', 'filename': 'index.html'}, }) eventlet.wsgi.server(eventlet.listen(('', 8000)), app) class engineio.ASGIApp(engineio_server, other_asgi_app=None, static_files=None, engineio_path='engine.io', on_startup=None, on_shutdown=None) ASGI Engine.IO. Engine.IO . HTTP ASGI . o engineio_server -- Engine.IO. engineio.AsyncServer . o static_files -- . . o other_asgi_app -- ASGI . o engineio_path -- (endpoint) Engine.IO . . None Engine.IO . None static_files other_asgi_app . o on_startup -- coroutine o on_shutdown -- coroutine : import engineio import uvicorn eio = engineio.AsyncServer() app = engineio.ASGIApp(eio, static_files={ '/': {'content_type': 'text/html', 'filename': 'index.html'}, '/index.html': {'content_type': 'text/html', 'filename': 'index.html'}, }) uvicorn.run(app, '127.0.0.1', 5000) async not_found(receive, send) 404 Not Found . class engineio.Middleware(engineio_app, wsgi_app=None, engineio_path='engine.io') WSGIApp . o <> o <> o <> Miguel Grinberg (COPYRIGHT) 2018, Miguel Grinberg July 18, 2026 PYTHON-ENGINEIO(1)