A Super Fast Async Python Web Framework with a Rust runtime.
npx @tessl/cli install tessl/pypi-robyn@0.72.0A 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.
pip install robynfrom robyn import Robyn, Request, ResponseCommon 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
)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)Robyn's architecture combines Python's simplicity with Rust's performance:
Robyn class serves as the main application instance, managing routing, middleware, and server lifecycleMain 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): ...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): ...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): ...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): ...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): ...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): ...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 constantsTemplate 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: ...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: ...@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"