Robyn Documentation

pip install robyn

from robyn import Robyn

app = Robyn(__file__)

@app.get("/")
async def index(request):
    return "Hello, World!"

app.start(port=8080)

robyn/
├── src/                    # Rust source code
│   ├── lib.rs              # PyO3 module entry point
│   ├── server.rs           # Main HTTP server implementation
│   ├── types/              # Request, Response, Headers, Cookie types
│   ├── routers/            # HTTP, WebSocket, middleware routers
│   ├── executors/          # Route execution handlers
│   └── websockets/         # WebSocket implementation
├── robyn/                  # Python package
│   ├── __init__.py         # Main Robyn and SubRouter classes
│   ├── router.py           # Python router implementation
│   ├── authentication.py   # AuthenticationHandler
│   ├── dependency_injection.py
│   ├── openapi.py          # OpenAPI generation
│   ├── mcp.py              # MCP protocol support
│   ├── ai.py               # AI agent support
│   ├── responses.py        # Response helpers (serve_file, html, SSE)
│   ├── ws.py               # WebSocket class
│   └── robyn.pyi           # Type stubs
├── integration_tests/      # Integration test suite
├── unit_tests/             # Unit test suite
├── docs_src/               # Documentation (Next.js)
├── granian/                # Bundled Granian server (fork)
└── examples/               # Example applications

from robyn import Robyn, SubRouter

app = Robyn(__file__)
api = SubRouter(__file__, prefix="/api")

@api.get("/users")
def get_users(request):
    return {"users": []}

app.include_router(api)

request.method      # HTTP method
request.url         # Url object (scheme, host, path)
request.headers     # Headers dict-like
request.query_params # QueryParams
request.path_params  # Dict of URL params
request.body        # Raw bytes
request.json()      # Parse JSON body
request.form_data   # Multipart form data
request.ip_addr     # Client IP
request.identity    # Identity (if authenticated)

from robyn import Response

Response(
    status_code=200,
    headers={"Content-Type": "application/json"},
    description="body content"  # or body bytes
)

@app.get("/path")
@app.post("/path")
@app.put("/path")
@app.delete("/path")
@app.patch("/path")
@app.head("/path")
@app.options("/path")

@app.before_request("/path")  # Middleware before
@app.after_request("/path")   # Middleware after

@app.startup_handler         # Server startup
@app.shutdown_handler        # Server shutdown

from robyn import WebSocketDisconnect

@app.websocket("/ws")
async def handler(websocket):
    try:
        while True:
            msg = await websocket.receive_text()
            await websocket.send_text(f"Echo: {msg}")
    except WebSocketDisconnect:
        pass

@handler.on_connect
def on_connect(websocket):
    return "Connected"

@handler.on_close
def on_close(websocket):
    return "Closed"

from typing import List, Optional

# HTTP: path params + query params with type coercion
@app.get("/items/:id")
async def get_item(id: int, q: str, page: int = 1):
    return {"id": id, "q": q, "page": page}

# Optional, List, and bool params
@app.get("/search")
def search(name: str, tags: List[str], active: bool = False, age: Optional[int] = None):
    return {"name": name, "tags": tags, "active": active, "age": age}

# WebSocket: typed query params on handler and callbacks
@app.websocket("/ws")
async def handler(websocket, room: str = "default", page: int = 1):
    while True:
        msg = await websocket.receive_text()
        await websocket.send_text(f"room={room} page={page} msg={msg}")

@handler.on_connect
def on_connect(websocket, room: str = "default"):
    return f"connected to {room}"

@app.mcp.resource("time://current")
def get_time():
    return datetime.now().isoformat()

@app.mcp.tool(name="calc", description="Calculate", input_schema={...})
def calculate(args):
    return eval(args["expression"])

@app.mcp.prompt(name="explain", description="Explain code", arguments=[...])
def explain_prompt(args):
    return f"Please explain: {args['code']}"

python app.py                     # Start server
python app.py --dev               # Development mode (hot reload)
python app.py --processes 4       # Multi-process
python app.py --workers 2         # Workers per process
python app.py --log-level DEBUG   # Log level
python app.py --open-browser      # Open browser on start
python app.py --create            # Create new project scaffold
python app.py --docs              # Open documentation

# Clone
git clone https://github.com/sparckles/robyn.git
cd robyn

# Virtual environment
python3 -m venv .venv && source .venv/bin/activate

# Install tools
pip install pre-commit poetry maturin

# Install dependencies
poetry install --with dev --with test

# Build Rust extension
maturin develop

# Run tests
pytest