SIO-AsyncAPI

SIO-AsyncAPI is a Python library built on the top of Flask-SocketIO and driven by AsyncAPI. It allows you to generate an AsyncAPI specification from your SocketIO server and validate messages against it.

Similar to FastAPI, SIO-AsyncAPI allows you to define your SocketIO server using Python type annotations and Pydantic models. It also provides a way to generate an AsyncAPI specification from your SocketIO server.

SIO-AsyncAPI now supports both Pydantic 1.10+ and 2.x, emits AsyncAPI 3.1, and keeps Socket.IO ACKs in the custom x-ack extension.

Installation

pip install sio_asyncapi

Basic Example

# examples/simple.py

from flask import Flask
from sio_asyncapi import AsyncAPISocketIO, ResponseValidationError, RequestValidationError
from pydantic import BaseModel, Field, EmailStr
from typing import Optional
import logging
logger = logging.getLogger(__name__)

app = Flask(__name__)

socketio = AsyncAPISocketIO(
    app,
    validate=True,
    generate_docs=True,
    version="1.0.0",
    title="Demo",
    description="Demo Server",
    server_url="http://localhost:5000",
    server_name="DEMO_SIO",
)


class UserSignUpRequest(BaseModel):
    """Request model for user sign up"""
    email: EmailStr = Field(..., description="User email", example="bob@gmail.com")
    password: str = Field(..., description="User password", example="123456")


class UserSignUpResponse(BaseModel):
    """Response model for user sign up"""
    success: bool = Field(True, description="Success status")
    error: Optional[str] = Field( None, description="Error message if any",
        example="Invalid request")


@socketio.on("user_sign_up", get_from_typehint=True)
def user_sign_up(request: UserSignUpRequest) -> UserSignUpResponse:
    """User sign up"""
    _ = request
    return UserSignUpResponse(success=True, error=None)

@socketio.on_error_default
def default_error_handler(e: Exception):
    """
    Default error handler. It called if no other error handler defined.
    Handles RequestValidationError and ResponseValidationError errors.
    """
    if isinstance(e, RequestValidationError):
        logger.error(f"Request validation error: {e}")
        return {"success": False, "error": str(e)}
    elif isinstance(e, ResponseValidationError):
        logger.critical(f"Response validation error: {e}")
        raise e
    else:
        logger.critical(f"Unknown error: {e}")
        raise e

if __name__ == '__main__':
    socketio.run(app, debug=True)

# import pathlib
# if __name__ == "__main__":
#     path = pathlib.Path(__file__).parent / "simple.yml"
#     doc_str = socketio.asyncapi_doc.get_yaml()
#     with open(path, "w") as f:
#         f.write(doc_str)
#     print(doc_str)

Here is how validation error looks like in FireCamp:

In order to get the AsyncAPI specification from your SocketIO server instead of running the server, you can do the following: You can also get a compact agent-friendly event catalog with socketio.get_agent_schema() or socketio.get_agent_schema_json(). See examples/agentic_doc_example.py for a complete example that writes both exports to disk.

import pathlib
if __name__ == "__main__":
    path = pathlib.Path(__file__).parent / "simple.yml"
    doc_str = socketio.asyncapi_doc.get_yaml()
    with open(path, "w") as f:
        f.write(doc_str)
    print(doc_str)

Example of the AsyncAPI specification generated from the above example:

# examples/simple.yml

asyncapi: 3.1.0
info:
  title: Demo
  version: 1.0.0
  description: 'Demo Server

    <br/> This specification targets AsyncAPI 3.1 and keeps Socket.IO ACK values in
    the custom `x-ack` message extension.

    Socket.IO-specific transport details may still require application-level interpretation.

    '
servers:
  DEMO_SIO:
    host: localhost:5000
    protocol: socketio
channels:
  root:
    address: /
    messages:
      User_Sign_Up:
        $ref: '#/components/messages/User_Sign_Up'
operations:
  receive_user_sign_up:
    action: receive
    channel:
      $ref: '#/channels/root'
    messages:
    - $ref: '#/channels/root/messages/User_Sign_Up'
    description: User sign up
components:
  messages:
    User_Sign_Up:
      name: user_sign_up
      description: User sign up
      payload:
        $ref: '#/components/schemas/UserSignUpRequest'
      x-ack:
        $ref: '#/components/schemas/UserSignUpResponse'
  schemas:
    NoSpec:
      description: Specification is not provided
    UserSignUpRequest:
      title: UserSignUpRequest
      description: Request model for user sign up
      type: object
      properties:
        email:
          title: Email
          description: User email
          example: bob@gmail.com
          type: string
          format: email
        password:
          title: Password
          description: User password
          example: '123456'
          type: string
      required:
      - email
      - password
    UserSignUpResponse:
      title: UserSignUpResponse
      description: Response model for user sign up
      type: object
      properties:
        success:
          title: Success
          description: Success status
          default: true
          type: boolean
        error:
          title: Error
          description: Error message if any
          example: Invalid request
          type: string

Rendered version of the above AsyncAPI specification:

Converting from Flask-SocketIO to SIO-AsyncAPI

SIO-AsyncAPI is built on top of Flask-SocketIO and all unit tests of Flask-SocketIO are tested against SIO-AsyncAPI. If you converting your SocketIO server from Flask-SocketIO to SIO-AsyncAPI, you can be sure that your SocketIO server will work as expected. When converting your SocketIO server from Flask-SocketIO to SIO-AsyncAPI, it's as simple as changing the import statement:

# instead of `from flask_socketio import SocketIO`
from sio_asyncapi import AsyncAPISocketIO as SocketIO
...
# There are additional arguments that you can pass to the constructor of AsyncAPISocketIO
socketio = SocketIO(app)
...

Acknowledgements

Most of the implementation follows research done by Dimitrios Dedoussis (https://www.asyncapi.com/blog/socketio-part2) and uses some Pydantic models from here

Missing Features

SIO-AsyncAPI is still in its early stages and there are some features that are not yet implemented. If you are interested in contributing to SIO-AsyncAPI any contribution is welcome. Here is the list of missing features:

• Support of AsycnAPI documentation and validation for emit messages
• Support of Flask-SocketIO namespaces and rooms
• Authentication and security auto documentation
• connect and disconnect handlers auto documentation