or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

authentication.mdcore-app.mdexceptions.mdindex.mdmcp.mdopenapi.mdrequest-response.mdstatus-codes.mdtemplating.mdwebsocket.md
tile.json

tessl/pypi-robyn

A Super Fast Async Python Web Framework with a Rust runtime.

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/robyn@0.72.x

To install, run

npx @tessl/cli install tessl/pypi-robyn@0.72.0

index.mddocs/

Robyn

A Super Fast Async Python Web Framework with a Rust runtime. Robyn combines Python's ease of use with Rust's speed and safety, providing a simple, FastAPI-like API for building high-performance web applications and APIs with automatic OpenAPI generation, WebSocket support, middleware capabilities, and built-in features like hot reloading, dependency injection, and direct Rust integration.

Package Information

  • Package Name: robyn
  • Package Type: pypi
  • Language: Python
  • Installation: pip install robyn

Core Imports

from robyn import Robyn, Request, Response

Common imports for building web applications:

from robyn import (
    Robyn, Request, Response, SubRouter,
    serve_file, serve_html, html,
    StreamingResponse, SSEResponse,
    WebSocket, WebSocketConnector,
    AuthenticationHandler, BearerGetter,
    status_codes, jsonify, ALLOW_CORS
)

Basic Usage

from robyn import Robyn

# Create app instance
app = Robyn(__file__)

# Define a simple route
@app.get("/")
def index(request):
    return "Hello, World!"

# Define a route with path parameters
@app.get("/user/<user_id>")
def get_user(request):
    user_id = request.path_params["user_id"]
    return f"User ID: {user_id}"

# Define a POST route with JSON handling
@app.post("/users")
def create_user(request):
    user_data = request.json()
    return {"message": "User created", "data": user_data}

# Start the server
if __name__ == "__main__":
    app.start(host="0.0.0.0", port=8080)

Architecture

Robyn's architecture combines Python's simplicity with Rust's performance:

  • Core Application: The Robyn class serves as the main application instance, managing routing, middleware, and server lifecycle
  • Request/Response Pipeline: Built on Rust runtime for high-performance HTTP handling with Python-friendly objects
  • Async Support: Full async/await support with both sync and async route handlers
  • Middleware System: Before/after request middleware with endpoint-specific or global scope
  • Router System: Hierarchical routing with sub-routers for organizing large applications
  • WebSocket Integration: Real-time communication with event-driven WebSocket handling
  • Authentication: Pluggable authentication system with bearer token support
  • OpenAPI Generation: Automatic API documentation generation with Swagger UI

Capabilities

Core Application

Main application class and routing functionality for creating web applications with HTTP methods, middleware, static file serving, and lifecycle management.

class Robyn:
    def __init__(
        self, 
        file_object: str, 
        config: Config = Config(), 
        openapi_file_path: Optional[str] = None,
        openapi: Optional[OpenAPI] = None,
        dependencies: DependencyMap = DependencyMap()
    ): ...
    
    def start(
        self,
        host: str = "127.0.0.1",
        port: int = 8080,
        _check_port: bool = True,
        client_timeout: int = 30,
        keep_alive_timeout: int = 20
    ): ...
    
    def get(self, endpoint: str, auth_required: bool = False): ...
    def post(self, endpoint: str, auth_required: bool = False): ...
    def put(self, endpoint: str, auth_required: bool = False): ...
    def delete(self, endpoint: str, auth_required: bool = False): ...

Core Application

Request and Response Handling

HTTP request and response objects with headers, query parameters, form data, file uploads, and response utilities including HTML, file serving, and streaming responses.

@dataclass
class Request:
    query_params: QueryParams
    headers: Headers
    path_params: dict[str, str]
    body: Union[str, bytes]
    method: str
    url: Url
    form_data: dict[str, str]
    files: dict[str, bytes]
    ip_addr: Optional[str]
    identity: Optional[Identity]
    
    def json(self) -> dict: ...

@dataclass  
class Response:
    status_code: int
    headers: Union[Headers, dict]
    description: Union[str, bytes]
    response_type: Optional[str]
    file_path: Optional[str]
    
    def set_cookie(self, key: str, value: str): ...

Request and Response

WebSocket Support

Real-time communication with WebSocket connections, event handling (connect, message, close), broadcasting, and direct messaging capabilities.

class WebSocket:
    def __init__(
        self, 
        robyn_object: "Robyn", 
        endpoint: str, 
        config: Config = Config(),
        dependencies: DependencyMap = DependencyMap()
    ): ...
    
    def on(self, type: str): ...

class WebSocketConnector:
    id: str
    query_params: QueryParams
    
    def async_broadcast(self, message: str): ...
    def async_send_to(self, sender_id: str, message: str): ...
    def sync_broadcast(self, message: str): ...
    def sync_send_to(self, sender_id: str, message: str): ...

WebSocket

Authentication

Pluggable authentication system with bearer token support, custom authentication handlers, and identity management for securing routes.

class AuthenticationHandler:
    def __init__(self, token_getter: TokenGetter): ...
    def authenticate(self, request: Request) -> Optional[Identity]: ...
    @property
    def unauthorized_response(self) -> Response: ...

class BearerGetter:
    def get_token(self, request: Request) -> Optional[str]: ...
    def set_token(self, request: Request, token: str): ...

Authentication

OpenAPI Integration

Automatic API documentation generation with Swagger UI, route documentation, and custom OpenAPI specification support.

class OpenAPI:
    def add_openapi_path_obj(
        self, 
        route_type, 
        endpoint, 
        openapi_name, 
        openapi_tags, 
        handler
    ): ...
    def get_openapi_docs_page(self) -> str: ...
    def get_openapi_config(self) -> dict: ...
    def override_openapi(self, openapi_json_spec_path: str): ...

OpenAPI

Model Context Protocol (MCP) Integration

AI agent integration through MCP interface with resource registration, tool definitions, and prompt templates for building AI-enabled web applications.

class MCPApp:
    def resource(self, uri: str, name: str, description: str): ...
    def tool(self, name: str, description: str, input_schema: dict): ...
    def prompt(self, name: str, description: str): ...

MCP Integration

HTTP Status Codes

Comprehensive HTTP status code constants for proper response status handling, including informational, success, redirection, client error, and server error codes.

# Standard HTTP status codes
HTTP_200_OK = 200
HTTP_201_CREATED = 201
HTTP_400_BAD_REQUEST = 400
HTTP_401_UNAUTHORIZED = 401
HTTP_404_NOT_FOUND = 404
HTTP_500_INTERNAL_SERVER_ERROR = 500
# ... and 67 more status code constants

HTTP Status Codes

Template Rendering

Template rendering system with Jinja2 support and custom template interfaces for building dynamic web applications.

class TemplateInterface(ABC):
    @abstractmethod
    def render_template(self, *args, **kwargs) -> Response: ...

class JinjaTemplate(TemplateInterface):
    def __init__(self, directory: str, encoding: str = "utf-8", followlinks: bool = False): ...
    def render_template(self, template_name: str, **kwargs) -> Response: ...

Templating

Exception Handling

Exception classes for handling HTTP and WebSocket errors with proper status codes and error messages.

class HTTPException(Exception):
    def __init__(self, status_code: int, detail: str | None = None) -> None: ...

class WebSocketException(Exception):
    def __init__(self, code: int, reason: str | None = None) -> None: ...

Exception Handling

Types

@dataclass
class Identity:
    claims: dict[str, str]

@dataclass  
class Url:
    scheme: str
    host: str
    path: str

class Headers:
    def __init__(self, default_headers: Optional[dict]): ...
    def get(self, key: str): ...
    def set(self, key: str, value: str): ...
    def contains(self, key: str) -> bool: ...

class QueryParams:
    def get(self, key: str, default: Optional[str] = None): ...
    def get_all(self, key: str): ...
    def set(self, key: str, value: str): ...
    def to_dict(self) -> dict: ...

from enum import Enum

class HttpMethod(Enum):
    GET = "GET"
    POST = "POST" 
    PUT = "PUT"
    DELETE = "DELETE"
    PATCH = "PATCH"
    OPTIONS = "OPTIONS"
    HEAD = "HEAD"
    TRACE = "TRACE"
    CONNECT = "CONNECT"

class Events(Enum):
    STARTUP = "startup"
    SHUTDOWN = "shutdown"