Synchronous Server

class Server:
    def __init__(
        self,
        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
    ):
        """
        Initialize Engine.IO server.

        Args:
            async_mode (str, optional): Async model ('threading', 'eventlet', 'gevent', 'gevent_uwsgi')
            ping_interval (int|tuple): Ping interval in seconds, default 25
            ping_timeout (int): Ping timeout in seconds, default 20
            max_http_buffer_size (int): Max message size in bytes, default 1,000,000
            allow_upgrades (bool): Allow transport upgrades, default True
            http_compression (bool): Enable HTTP compression, default True
            compression_threshold (int): Compression threshold in bytes, default 1024
            cookie (str|dict|None): Cookie configuration
            cors_allowed_origins (str|list|callable): CORS allowed origins
            cors_credentials (bool): Allow credentials in CORS, default True
            logger (bool|Logger): Logging configuration, default False
            json (module): Alternative JSON module
            async_handlers (bool): Run handlers asynchronously, default True
            monitor_clients (bool): Monitor client connections, default True
            transports (list): Allowed transports, default ['polling', 'websocket']
        """

def send(self, sid, data):
    """
    Send a message to a client.

    Args:
        sid (str): Session ID of the client
        data (any): Message data to send

    Raises:
        SocketIsClosedError: If client socket is closed
    """

def send_packet(self, sid, pkt):
    """
    Send a raw packet to a client.

    Args:
        sid (str): Session ID of the client
        pkt (Packet): Raw packet object to send

    Raises:
        SocketIsClosedError: If client socket is closed
    """

def get_session(self, sid):
    """
    Return the user session for a client.

    Args:
        sid (str): Session ID of the client

    Returns:
        dict: User session data

    Raises:
        KeyError: If session does not exist
    """

def save_session(self, sid, session):
    """
    Store the user session for a client.

    Args:
        sid (str): Session ID of the client
        session (dict): Session data to store
    """

def session(self, sid):
    """
    Return user session with context manager syntax.

    Args:
        sid (str): Session ID of the client

    Returns:
        ContextManager: Session context manager for automatic save
    """

# Direct session management
session_data = eio.get_session(sid)
session_data['user_id'] = 123
eio.save_session(sid, session_data)

# Context manager approach
with eio.session(sid) as session:
    session['user_id'] = 123
    session['preferences'] = {'theme': 'dark'}

def disconnect(self, sid=None):
    """
    Disconnect a client or all clients.

    Args:
        sid (str, optional): Session ID to disconnect. If None, disconnects all clients
    """

def transport(self, sid):
    """
    Return the transport name for a client.

    Args:
        sid (str): Session ID of the client

    Returns:
        str: Transport name ('polling' or 'websocket')

    Raises:
        KeyError: If client not found
    """

def on(self, event, handler=None):
    """
    Register an event handler.

    Args:
        event (str): Event name ('connect', 'message', 'disconnect')
        handler (callable, optional): Event handler function

    Returns:
        callable: Decorator function if handler not provided
    """

# Decorator syntax
@eio.on('connect')
def on_connect(sid, environ):
    print(f'Client {sid} connected')

@eio.on('message')
def on_message(sid, data):
    print(f'Received: {data}')
    eio.send(sid, f'Echo: {data}')

@eio.on('disconnect')
def on_disconnect(sid):
    print(f'Client {sid} disconnected')

# Method syntax
def handle_connect(sid, environ):
    print(f'Client {sid} connected')

eio.on('connect', handle_connect)

def handle_request(self, environ, start_response):
    """
    Handle an HTTP request (WSGI entry point).

    Args:
        environ (dict): WSGI environment dictionary
        start_response (callable): WSGI start_response callable

    Returns:
        iterable: WSGI response iterable
    """

def start_background_task(self, target, *args, **kwargs):
    """
    Start a background task.

    Args:
        target (callable): Task function to run
        *args: Arguments for the task function
        **kwargs: Keyword arguments for the task function

    Returns:
        Task object specific to the async mode
    """

def sleep(self, seconds=0):
    """
    Sleep using the appropriate async model.

    Args:
        seconds (float): Sleep duration in seconds
    """

def create_queue(self, *args, **kwargs):
    """
    Create a queue object appropriate for the async mode.

    Returns:
        Queue object
    """

def get_queue_empty_exception(self):
    """
    Return the queue empty exception for the async mode.

    Returns:
        Exception: Queue empty exception class
    """

def create_event(self, *args, **kwargs):
    """
    Create an event object appropriate for the async mode.

    Returns:
        Event object
    """

def generate_id(self):
    """
    Generate a unique session ID.

    Returns:
        str: Unique session identifier
    """

def shutdown(self):
    """
    Stop all background tasks and clean up resources.
    """

class Server.reason:
    SERVER_DISCONNECT = 'server disconnect'
    CLIENT_DISCONNECT = 'client disconnect'
    PING_TIMEOUT = 'ping timeout'
    TRANSPORT_CLOSE = 'transport close'
    TRANSPORT_ERROR = 'transport error'

# Available compression methods
compression_methods = ['gzip', 'deflate']

# Valid event names
event_names = ['connect', 'disconnect', 'message']

# Supported transport protocols
valid_transports = ['polling', 'websocket']