### Define a Basic Robyn API

Source: https://github.com/sparckles/robyn/blob/main/README.md

Create a simple web application by defining routes and starting the Robyn server. This example sets up a GET route for the root path that returns 'Hello, world!'.

```python
from robyn import Robyn

app = Robyn(__file__)

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

app.start(port=8080)
```

--------------------------------

### Quick Start: Chat Endpoint with AI Agent

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/ai.mdx

A basic example demonstrating how to set up a chat endpoint using Robyn's AI agent and in-memory storage for conversation history.

```python
from robyn import Robyn
from robyn.ai import agent, memory

app = Robyn(__file__)

# Create memory instance
mem = memory(provider="inmemory", user_id="user123")

# Create agent with memory
chat_agent = agent(runner="simple", memory=mem)

@app.get("/chat")
async def chat_endpoint(request):
    query = request.query_params.get("q", [""])[0]
    if not query:
        return {"error": "Query required"}
    
    # Run agent with conversation history
    result = await chat_agent.run(query, history=True)
    return result
```

--------------------------------

### Install dependencies

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/graphql-support.mdx

Install Robyn and strawberry-graphql packages.

```bash
pip install robyn strawberry-graphql
```

--------------------------------

### Setup Virtual Environment

Source: https://github.com/sparckles/robyn/blob/main/README.md

Create and activate a Python virtual environment.

```bash
python3 -m venv .venv
source .venv/bin/activate
```

--------------------------------

### Install Local Wheel

Source: https://github.com/sparckles/robyn/blob/main/README.md

Install a specific local build of Robyn into another project.

```bash
pip install -e path/to/robyn/target/wheels/robyn-<version>-<python_version>-<platform>.whl
```

```bash
pip install -e /repos/Robyn/target/wheels/robyn-0.63.0-cp312-cp312-macosx_10_15_universal2.whl
```

--------------------------------

### Advanced Monitoring Setup in Robyn

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/scaling.mdx

This snippet demonstrates setting up comprehensive application and system metrics collection in a Robyn application. It includes request monitoring, response time tracking, status code counts, and system resource usage. Ensure `psutil` is installed (`pip install psutil`).

```python
from robyn import Robyn
import time
import psutil
import logging
import threading
from collections import defaultdict, deque

app = Robyn(__file__)

# Configure structured logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler('app.log')
    ]
)
logger = logging.getLogger(__name__)

# Advanced metrics collection
class MetricsCollector:
    def __init__(self):
        self.request_count = 0
        self.total_response_time = 0
        self.status_codes = defaultdict(int)
        self.endpoint_stats = defaultdict(lambda: {"count": 0, "total_time": 0})
        self.response_times = deque(maxlen=1000)  # Last 1000 requests
        self.lock = threading.Lock()
    
    def record_request(self, method, path, status_code, duration):
        with self.lock:
            self.request_count += 1
            self.total_response_time += duration
            self.status_codes[status_code] += 1
            endpoint_key = f"{method} {path}"
            self.endpoint_stats[endpoint_key]["count"] += 1
            self.endpoint_stats[endpoint_key]["total_time"] += duration
            self.response_times.append(duration)
    
    def get_metrics(self):
        with self.lock:
            avg_response_time = self.total_response_time / max(self.request_count, 1)
            p95_response_time = sorted(self.response_times)[int(len(self.response_times) * 0.95)] if self.response_times else 0
            
            return {
                "requests_total": self.request_count,
                "avg_response_time": avg_response_time,
                "p95_response_time": p95_response_time,
                "status_codes": dict(self.status_codes),
                "top_endpoints": dict(sorted(
                    self.endpoint_stats.items(),
                    key=lambda x: x[1]["count"],
                    reverse=True
                )[:10])
            }

metrics = MetricsCollector()
start_time = time.time()

@app.before_request
def monitor_request_start(request):
    request.start_time = time.time()
    return request

@app.after_request
def monitor_request_end(request, response):
    duration = time.time() - request.start_time
    status_code = getattr(response, 'status_code', 200)
    
    # Record metrics
    metrics.record_request(request.method, request.url.path, status_code, duration)
    
    # Log slow requests
    if duration > 1.0:
        logger.warning(
            f"Slow request: {request.method} {request.url.path} - "
            f"{duration:.3f}s (status: {status_code})"
        )
    
    # Add response headers
    response.headers["X-Response-Time"] = f"{duration:.3f}s"
    response.headers["X-Request-ID"] = str(time.time_ns())
    return response

# Health endpoint with detailed status
@app.get("/health", const=True)
def health_check():
    return {
        "status": "healthy",
        "version": "1.0.0",
        "uptime_seconds": time.time() - start_time
    }

# Comprehensive metrics endpoint
@app.get("/metrics")
def get_metrics():
    cpu_percent = psutil.cpu_percent(interval=1)
    memory = psutil.virtual_memory()
    disk = psutil.disk_usage('/')
    
    app_metrics = metrics.get_metrics()
    
    return {
        "system": {
            "cpu_usage_percent": cpu_percent,
            "memory_usage_percent": memory.percent,
            "memory_available_mb": memory.available / 1024 / 1024,
            "disk_usage_percent": disk.percent,
            "load_average": psutil.getloadavg()
        },
        "application": app_metrics,
        "timestamp": time.time()
    }

# Readiness endpoint for k8s
@app.get("/ready")
def readiness_check():
    # Add your readiness checks here (DB connection, etc.)
    return {"ready": True}

```

--------------------------------

### Setup development environment

Source: https://github.com/sparckles/robyn/blob/main/docs_src/public/llms.txt

Commands to clone the repository, set up a virtual environment, and build the Rust extension.

```bash
# 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
```

--------------------------------

### Basic Application Setup

Source: https://context7.com/sparckles/robyn/llms.txt

Demonstrates how to create a Robyn application instance and define basic routes for handling HTTP requests.

```APIDOC
## Basic Application Setup

Create a Robyn application instance and define routes using decorator syntax for handling HTTP requests.

```python
from robyn import Robyn, Request

app = Robyn(__file__)

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

@app.get("/health", const=True)
def health_check():
    # const=True caches response in Rust for maximum performance
    return {"status": "healthy", "version": "1.0.0"}

if __name__ == "__main__":
    app.start(port=8080, host="0.0.0.0")
```
```

--------------------------------

### Install Dependencies

Source: https://github.com/sparckles/robyn/blob/main/README.md

Install required build tools and development packages.

```bash
pip install pre-commit poetry maturin
```

```bash
poetry install --with dev --with test
```

--------------------------------

### Sample Robyn.env File Configuration

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/robyn_env.mdx

Provides an example of a robyn.env file with various configuration settings.

```bash
ROBYN_PORT=8080
ROBYN_HOST=127.0.0.1
RANDOM_ENV=123
ROBYN_BROWSER_OPEN=True
ROBYN_DEV_MODE=True
ROBYN_MAX_PAYLOAD_SIZE=1000000
```

--------------------------------

### Install Robyn AI

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/ai.mdx

Install the Robyn framework, which includes the AI features, using pip.

```bash
pip install robyn
```

--------------------------------

### Start Robyn Server

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/architecture.mdx

Starts the Robyn web server. It configures host and port from environment variables or provided arguments, checks for port availability, and initiates the server processes. Ensure necessary configurations are set before calling this function.

```python
def start(self, host: str = "127.0.0.1", port: int = 8080, _check_port: bool = True):
        """
        Starts the server

        :param host str: represents the host at which the server is listening
        :param port int: represents the port number at which the server is listening
        :param _check_port bool: represents if the port should be checked if it is already in use
        """

        host = os.getenv("ROBYN_HOST", host)
        port = int(os.getenv("ROBYN_PORT", port))
        open_browser = bool(os.getenv("ROBYN_BROWSER_OPEN", self.config.open_browser))

        if _check_port:
            while self.is_port_in_use(port):
                logger.error("Port %s is already in use. Please use a different port.", port)
                try:
                    port = int(input("Enter a different port: "))
                except Exception:
                    logger.error("Invalid port number. Please enter a valid port number.")
                    continue

        logger.info("Robyn version: %s", __version__)
        logger.info("Starting server at http://%s:%s", host, port)

        mp.allow_connection_pickling()

        run_processes(
            host,
            port,
            self.directories,
            self.request_headers,
            self.router.get_routes(),
            self.middleware_router.get_global_middlewares(),
            self.middleware_router.get_route_middlewares(),
            self.web_socket_router.get_routes(),
            self.event_handlers,
            self.config.workers,
            self.config.processes,
            self.response_headers,
            open_browser,
        )
```

--------------------------------

### Install Robyn with templating extension using pip

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/index.mdx

Install Robyn along with the templating extension using pip.

```bash
pip install "robyn[templating]"
```

--------------------------------

### Performance Testing with wrk and Python

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/scaling.mdx

Use these examples to benchmark application performance under different process and worker configurations.

```bash
# Install wrk for load testing
# macOS: brew install wrk
# Ubuntu: sudo apt install wrk

# Test baseline performance
wrk -t4 -c100 -d30s http://localhost:8080/health

# Test with different configurations
# Configuration 1: Single process
python app.py --processes 1 --workers 1 &
wrk -t4 -c100 -d30s http://localhost:8080/api/endpoint

# Configuration 2: Multi-process
python app.py --processes 4 --workers 2 &
wrk -t4 -c100 -d30s http://localhost:8080/api/endpoint

# Configuration 3: Fast mode
python app.py --fast &
wrk -t4 -c100 -d30s http://localhost:8080/api/endpoint

# Compare results and choose the best configuration
```

```python
# Python script for automated testing
import subprocess
import time
import requests

def test_configuration(processes, workers, duration=30):
    # Start server
    cmd = f"python app.py --processes {processes} --workers {workers}"
    server = subprocess.Popen(cmd.split())
    time.sleep(2)  # Wait for startup
    
    try:
        # Run load test
        wrk_cmd = f"wrk -t4 -c100 -d{duration}s http://localhost:8080/health"
        result = subprocess.run(wrk_cmd.split(), capture_output=True, text=True)
        return result.stdout
    finally:
        server.terminate()
        time.sleep(1)

# Test different configurations
configs = [(1, 1), (2, 2), (4, 1), (4, 2), (8, 1)]
for processes, workers in configs:
    print(f"\nTesting: {processes} processes, {workers} workers")
    result = test_configuration(processes, workers)
    print(result)
```

--------------------------------

### Start Robyn Server in Rust

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/architecture.mdx

Initializes the server, sets up the event loop, and prepares routing structures for the Rust-based request handling.

```rust
    pub fn start(
        &mut self,
        py: Python,
        socket: &PyCell<SocketHeld>,
        workers: usize,
    ) -> PyResult<()> {
        pyo3_log::init();

        if STARTED
            .compare_exchange(false, true, SeqCst, Relaxed)
            .is_err()
        {
            debug!("Robyn is already running...");
            return Ok(());
        }

        let raw_socket = socket.try_borrow_mut()?.get_socket();

        let router = self.router.clone();
        let const_router = self.const_router.clone();
        let middleware_router = self.middleware_router.clone();
        let web_socket_router = self.websocket_router.clone();
        let global_request_headers = self.global_request_headers.clone();
        let global_response_headers = self.global_response_headers.clone();
        let directories = self.directories.clone();

        let asyncio = py.import("asyncio")?;
        let event_loop = asyncio.call_method0("new_event_loop")?;
        asyncio.call_method1("set_event_loop", (event_loop,))?;

        let startup_handler = self.startup_handler.clone();
        let shutdown_handler = self.shutdown_handler.clone();
```

--------------------------------

### Install Robyn with Pip

Source: https://github.com/sparckles/robyn/blob/main/README.md

Use Pip to install the Robyn framework. For optional features like Pydantic validation and Jinja2 templating, install with the 'all' extra.

```bash
pip install robyn
```

```bash
pip install "robyn[all]"
```

--------------------------------

### Install Robyn using Pip

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/example_app/index.mdx

Install Robyn within a virtual environment using pip. Ensure you have Python 3 installed.

```bash
python3 -m venv venv
source venv/bin/activate
pip install robyn
```

--------------------------------

### Basic Robyn App

Source: https://github.com/sparckles/robyn/blob/main/docs_src/public/llms.txt

A minimal Robyn application with a single GET route that returns 'Hello, World!'. This is a starting point for building web applications with Robyn.

```python
from robyn import Robyn

app = Robyn(__file__)

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

app.start(port=8080)
```

--------------------------------

### Install enhanced system monitoring dependencies

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/agents.mdx

Install the psutil package to enable enhanced system monitoring capabilities.

```bash
pip install psutil  # Enhanced system monitoring
```

--------------------------------

### Example HTML Template

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/templating.mdx

A sample HTML file structure compatible with Jinja2 template rendering.

```html
<!DOCTYPE html>
        <html lang="en">
        <head>
          <meta charset="utf-8">
          <title>Results</title>
        </head>

        <body>
         Hello {{ framework }}! You're using {{ templating_engine }}.
        </body>
```

--------------------------------

### Install Robyn with Pydantic support

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/pydantic.mdx

Use these commands to install Robyn with the necessary Pydantic dependencies.

```bash
pip install "robyn[pydantic]"
```

```bash
pip install "robyn[all]"
```

```bash
conda install robyn pydantic -c conda-forge
```

--------------------------------

### Basic Robyn Application Setup

Source: https://context7.com/sparckles/robyn/llms.txt

Set up a basic Robyn application and define routes using decorators. The `const=True` argument caches responses in Rust for maximum performance.

```python
from robyn import Robyn, Request

app = Robyn(__file__)

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

@app.get("/health", const=True)
def health_check():
    # const=True caches response in Rust for maximum performance
    return {"status": "healthy", "version": "1.0.0"}

if __name__ == "__main__":
    app.start(port=8080, host="0.0.0.0")
```

--------------------------------

### Full GraphQL application implementation

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/graphql-support.mdx

A complete example of a Robyn application configured with a GraphQL schema and endpoints.

```python
from typing import List, Optional
from robyn import Robyn, jsonify
import json

import dataclasses
import strawberry
import strawberry.utils.graphiql


@strawberry.type
class User:
  name: str


@strawberry.type
class Query:
  @strawberry.field
  def user(self) -> Optional[User]:
      return User(name="Hello")


schema = strawberry.Schema(Query)

app = Robyn(__file__)


@app.get("/", const=True)
async def get():
  return strawberry.utils.graphiql.get_graphiql_html()


@app.post("/")
async def post(request):
  body = request.json()
  query = body["query"]
  variables = body.get("variables", None)
  context_value = {"request": request}
  root_value = body.get("root_value", None)
  operation_name = body.get("operation_name", None)

  data = await schema.execute(
      query,
      variables,
      context_value,
      root_value,
      operation_name,
  )

  return jsonify(
      {
          "data": (data.data),
          **({"errors": data.errors} if data.errors else {}),
          **({"extensions": data.extensions} if data.extensions else {}),
      }
  )


if __name__ == "__main__":
  app.start(port=8080, host="0.0.0.0")
```

--------------------------------

### Build and Install Robyn

Source: https://github.com/sparckles/robyn/blob/main/README.md

Compile the Rust package for local development.

```bash
maturin develop
```

```bash
maturin develop --cargo-extra-args="--features=io-uring"
```

--------------------------------

### Implement User Management API with Robyn

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/getting_started.mdx

A complete example demonstrating sync and async handlers, path parameters, query parameters, and JSON request body processing.

```python
from robyn import Robyn, Request
import asyncio
import json
import time
from typing import Dict, Any

app = Robyn(__file__)

# In-memory storage for demo (use a real database in production)
users: Dict[str, Dict[str, Any]] = {
    "1": {"id": "1", "name": "Alice", "email": "alice@example.com", "created_at": "2024-01-01"},
    "2": {"id": "2", "name": "Bob", "email": "bob@example.com", "created_at": "2024-01-02"}
}

# Const route for health checks (cached in Rust for max performance)
@app.get("/health", const=True)
def health_check():
    return {"status": "healthy", "version": "1.0.0"}

# Async handler for database-like operations
@app.get("/users/:id")
async def get_user(path_params):
    user_id = path_params["id"]
    
    # Simulate async database lookup
    await asyncio.sleep(0.01)  # Simulate DB query time
    
    if user_id in users:
        return {"success": True, "user": users[user_id]}
    else:
        return {"success": False, "error": "User not found"}, 404

# Get all users with pagination
@app.get("/users")
async def list_users(query_params):
    page = int(query_params.get("page", "1"))
    limit = int(query_params.get("limit", "10"))
    
    # Simulate async operation
    await asyncio.sleep(0.01)
    
    user_list = list(users.values())
    start = (page - 1) * limit
    end = start + limit
    
    return {
        "users": user_list[start:end],
        "total": len(user_list),
        "page": page,
        "limit": limit
    }

# Create new user
@app.post("/users")
async def create_user(body):
    try:
        data = json.loads(body)
        
        # Validate required fields
        if not data.get("name") or not data.get("email"):
            return {"success": False, "error": "Name and email are required"}, 400
        
        # Generate new ID
        new_id = str(len(users) + 1)
        new_user = {
            "id": new_id,
            "name": data["name"],
            "email": data["email"],
            "created_at": time.strftime("%Y-%m-%d")
        }
        
        # Simulate async database save
        await asyncio.sleep(0.02)
        users[new_id] = new_user
        
        return {"success": True, "user": new_user}, 201
        
    except json.JSONDecodeError:
        return {"success": False, "error": "Invalid JSON"}, 400

# Sync handler for CPU-intensive operations
@app.post("/calculate")
def calculate_fibonacci(body):
    try:
        data = json.loads(body)
        n = data.get("number", 10)
        
        if n < 0 or n > 35:  # Prevent excessive computation
            return {"error": "Number must be between 0 and 35"}, 400
        
        # CPU-intensive calculation (runs in thread pool)
        def fib(x):
            if x <= 1:
                return x
            return fib(x-1) + fib(x-2)
        
        start_time = time.time()
        result = fib(n)
        calc_time = time.time() - start_time
        
        return {
            "input": n,
            "result": result,
            "calculation_time": f"{calc_time:.4f}s"
        }
        
    except json.JSONDecodeError:
        return {"error": "Invalid JSON"}, 400

if __name__ == "__main__":
    app.start(port=8080)
```

--------------------------------

### Complete Robyn AI Chat Application Example

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/ai.mdx

A comprehensive example showcasing the integration of Robyn, AI agents, and memory management, including endpoints for chat, memory retrieval, clearing, and adding messages.

```python
from robyn import Robyn
from robyn.ai import agent, memory

app = Robyn(__file__)

# Create memory with InMemory provider
mem = memory(
    provider="inmemory",
    user_id="guest"
)

# Create agent with memory
chat_agent = agent(runner="simple", memory=mem)

@app.get("/")
async def home():
    return {"message": "Robyn AI Chat API"}

@app.post("/chat")  
async def chat(request):
    """Chat with AI agent"""
    data = request.json()
    query = data.get("query", "")
    include_history = data.get("history", True)
    
    if not query:
        return {"error": "Query is required"}
    
    try:
        result = await chat_agent.run(query, history=include_history)
        return {
            "query": query,
            "response": result.get("response"),
            "history_included": include_history
        }
    except Exception as e:
        return {"error": str(e)}

@app.get("/memory")
async def get_memory():
    """Retrieve conversation history"""
    try:
        memories = await mem.get()
        return {"memories": memories, "count": len(memories)}
    except Exception as e:
        return {"error": str(e)}

@app.delete("/memory")
async def clear_memory():
    """Clear conversation history"""
    try:
        await mem.clear()
        return {"message": "Memory cleared"}
    except Exception as e:
        return {"error": str(e)}

@app.post("/memory")
async def add_memory(request):
    """Add message to memory"""
    data = request.json()
    message = data.get("message", "")
    metadata = data.get("metadata", {})
    
    if not message:
        return {"error": "Message is required"}
    
    try:
        await mem.add(message, metadata)
        return {"message": "Added to memory"}
    except Exception as e:
        return {"error": str(e)}

if __name__ == "__main__":
    app.start(host="127.0.0.1", port=8080)
```

--------------------------------

### Run MCP Tests with Examples

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/mcps.mdx

Execute unit, live, or all tests for the MCP functionality using command-line arguments. This is useful for verifying the integration and functionality.

```bash
# Unit tests
python examples/mcp.py test-unit

# Live tests
python examples/mcp.py test-live

# All tests
python examples/mcp.py test-all
```

--------------------------------

### Define GET route for GraphiQL

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/graphql-support.mdx

Serve the GraphiQL IDE using a GET route with const=True for performance.

```python
  @app.get("/", const=True)
  async def get():
  return strawberry.utils.graphiql.get_graphiql_html()
```

--------------------------------

### Timeout Configuration Examples

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/timeout_configuration.mdx

Various configurations for different operational scenarios including production, development, and load testing.

```python
# Minimal timeout configuration
app.start(client_timeout=30)
```

```python
# Optimized for high-traffic scenarios
app.start(
    host="0.0.0.0",
    port=8080,
    client_timeout=60,      # Allow longer processing time
    keep_alive_timeout=15   # Shorter keep-alive for faster turnover
)
```

```python
# Development-friendly settings
app.start(
    client_timeout=300,     # Long timeout for debugging
    keep_alive_timeout=60   # Longer keep-alive for testing
)
```

```python
# Optimized for load testing with tools like wrk
app.start(
    client_timeout=10,      # Quick timeouts
    keep_alive_timeout=5    # Fast connection turnover
)
```

--------------------------------

### Run the MCP server

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/agents.mdx

Execute the agent server script to start the MCP service.

```bash
python examples/agents.py
```

--------------------------------

### Run Robyn with Open Browser Option

Source: https://github.com/sparckles/robyn/blob/main/README.md

Start the Robyn application and automatically open a new browser window to the application's URL.

```bash
python3 app.py --open-browser
```

--------------------------------

### Configure Git Hooks

Source: https://github.com/sparckles/robyn/blob/main/README.md

Install pre-commit hooks to ensure code quality.

```bash
pre-commit install
```

--------------------------------

### Quick Start: Initialize Robyn App and Register MCP Endpoints

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/mcps.mdx

Initialize a Robyn application and register resource, tool, and prompt endpoints using the MCP decorator. This sets up your application to interact with AI models.

```python
from robyn import Robyn

app = Robyn(__file__)

@app.mcp.resource("echo://{message}")
def echo_resource(message: str) -> str:
    return f"Resource echo: {message}"

@app.mcp.tool()
def echo_tool(message: str) -> str:
    return f"Tool echo: {message}"

@app.mcp.prompt()
def echo_prompt(message: str) -> str:
    return f"Please process: {message}"

app.start()
```

--------------------------------

### Create a Synchronous Robyn Application

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/getting_started.mdx

This example demonstrates a basic synchronous request handler in Robyn. It's suitable for simple operations or CPU-bound tasks.

```python
from robyn import Robyn, Request

app = Robyn(__file__)

@app.get("/")
def h(request: Request):
    return "Hello, world"

app.start(port=8080, host="0.0.0.0") # host is optional, defaults to 127.0.0.1
```

--------------------------------

### Robyn Event Handlers

Source: https://github.com/sparckles/robyn/blob/main/llms.txt

Shows decorators for defining startup and shutdown handlers in Robyn, which are executed when the server starts or stops.

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

--------------------------------

### Route Ordering for Specificity

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/advanced_routing.mdx

Register more specific routes before general ones to ensure correct matching. The example demonstrates the correct order for user-related routes.

```python
# GOOD: Specific routes first
@app.get("/users/profile")
def get_current_user_profile():
    return {"profile": "current_user"}

@app.get("/users/settings")
def get_user_settings():
    return {"settings": "user_settings"}

@app.get("/users/:id")
def get_user(path_params):
    return {"user_id": path_params["id"]}

# BAD: This would never be reached
# @app.get("/users/:id")  # Registered first
# @app.get("/users/profile")  # Never matched!
```

--------------------------------

### Sample Project Directory Structure

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/robyn_env.mdx

Illustrates a typical project directory structure including the robyn.env file.

```bash
--project/
  --robyn.env
  --index.py
  ...
```

--------------------------------

### Robyn Decorators for Routing and Middleware

Source: https://github.com/sparckles/robyn/blob/main/docs_src/public/llms.txt

Examples of decorators used to define HTTP routes (GET, POST, PUT, DELETE, etc.), middleware (before/after request), and server lifecycle handlers (startup/shutdown).

```python
@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
```

--------------------------------

### Initialize Robyn app

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/graphql-support.mdx

Create the Robyn application instance.

```python
app = Robyn(__file__)
```

--------------------------------

### Install OpenAI Dependency

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/ai.mdx

Install the required openai package to resolve ImportErrors.

```bash
pip install openai
```

--------------------------------

### Install Robyn with Conda

Source: https://github.com/sparckles/robyn/blob/main/README.md

Install Robyn using conda-forge for environments managed with Conda.

```bash
conda install -c conda-forge robyn
```

--------------------------------

### Create a virtual environment

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/graphql-support.mdx

Initialize a new virtual environment for the project.

```bash
python3 -m venv venv
```

--------------------------------

### Initialize and Use TestClient

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/testing.mdx

Import TestClient from robyn.testing, pass your Robyn app to it, and begin making requests to test your routes.

```python
from robyn import Robyn
from robyn.testing import TestClient

app = Robyn(__file__)

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

client = TestClient(app)

def test_hello():
    response = client.get("/hello")
    assert response.status_code == 200
    assert response.text == "Hello, World!"
```

--------------------------------

### Install Robyn with conda

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/index.mdx

Use this command to install the base Robyn package using conda from the conda-forge channel.

```bash
conda install robyn -c conda-forge
```

--------------------------------

### Configure Basic Robyn App with OpenAPI

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/openapi.mdx

Initializes a Robyn application with custom OpenAPI metadata and defines routes with query and path parameters.

```python
from robyn.robyn import QueryParams

from robyn import Robyn, Request

app = Robyn(
    file_object=__file__,
    openapi=OpenAPI(
        info=OpenAPIInfo(
            title="Sample App",
            description="This is a sample server application.",
            termsOfService="https://example.com/terms/",
            version="1.0.0",
            contact=Contact(
                name="API Support",
                url="https://www.example.com/support",
                email="support@example.com",
            ),
            license=License(
                name="BSD2.0",
                url="https://opensource.org/license/bsd-2-clause",
            ),
            externalDocs=ExternalDocumentation(description="Find more info here", url="https://example.com/"),
            components=Components(),
        ),
    ),
)


@app.get("/")
async def welcome():
    """welcome endpoint"""
    return "hi"


class GetRequestParams(QueryParams):
    appointment_id: str
    year: int


@app.get("/api/v1/name", openapi_name="Name Route", openapi_tags=["Name"])
async def get(r: Request, query_params: GetRequestParams):
    """Get Name by ID"""
    return r.query_params


@app.delete("/users/:name", openapi_tags=["Name"])
async def delete(r: Request):
    """Delete Name by ID"""
    return r.path_params


if __name__ == "__main__":
    app.start()
```

--------------------------------

### Install Robyn with templating extension using conda

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/index.mdx

Install Robyn along with the templating extension using conda from the conda-forge channel.

```bash
conda install "robyn[templating]" -c conda-forge
```

--------------------------------

### GET /users/:id

Source: https://context7.com/sparckles/robyn/llms.txt

Retrieve a user by their unique ID.

```APIDOC
## GET /users/:id

### Description
Get user by ID.

### Method
GET

### Endpoint
/users/:id

### Parameters
#### Path Parameters
- **id** (str) - Required - The unique identifier of the user

### Response
#### Success Response (200)
- **user_id** (str) - The ID of the retrieved user
```

--------------------------------

### Clone Repository

Source: https://github.com/sparckles/robyn/blob/main/README.md

Initial step to download the source code.

```bash
git clone https://github.com/sparckles/Robyn.git
```

--------------------------------

### Navigate to the project directory

Source: https://github.com/sparckles/robyn/blob/main/CONTRIBUTING.md

Change the current working directory to the cloned project folder.

```bash
cd robyn
```

--------------------------------

### GET /search

Source: https://context7.com/sparckles/robyn/llms.txt

Search for items in the database using query parameters.

```APIDOC
## GET /search

### Description
Search for items in the database.

### Method
GET

### Endpoint
/search

### Parameters
#### Query Parameters
- **q** (str) - Required - Search query string
- **page** (int) - Required - Page number
- **limit** (int) - Required - Limit of results per page

### Response
#### Success Response (200)
- **results** (list) - List of search results
- **query** (str) - The search query string
```

--------------------------------

### Serve files and HTML with Robyn

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/response-objects.mdx

Use serve_file for attachments, serve_html for template files, or FileResponse for custom headers.

```python
from robyn import Robyn
from robyn.responses import serve_file

app = Robyn(__file__)

@app.get("/download")
def download():
    return serve_file("report.pdf")
```

```python
from robyn.responses import serve_html

@app.get("/page")
def page():
    return serve_html("templates/index.html")
```

```python
from robyn.responses import FileResponse
from robyn.robyn import Headers

@app.get("/custom-file")
def custom_file():
    return FileResponse(
        file_path="data/export.csv",
        status_code=200,
        headers=Headers({"Content-Type": "text/csv"}),
    )
```

--------------------------------

### GET /search

Source: https://github.com/sparckles/robyn/blob/main/docs_src/public/llms.txt

Searches for items based on name and tags, supporting optional and list parameters.

```APIDOC
## GET /search

### Description
Searches for items based on provided criteria. This endpoint showcases handling of required string, list, optional integer, and boolean parameters.

### Method
GET

### Endpoint
/search

### Parameters
#### Query Parameters
- **name** (str) - Required - The name to search for.
- **tags** (List[str]) - Required - A list of tags to filter by.
- **active** (bool) - Optional - A flag to indicate if the item is active, defaults to False.
- **age** (Optional[int]) - Optional - The age to filter by.

### Response
#### Success Response (200)
- **name** (str) - The name of the item.
- **tags** (List[str]) - The list of tags associated with the item.
- **active** (bool) - Indicates if the item is active.
- **age** (Optional[int]) - The age of the item.

### Response Example
```json
{
  "name": "example_item",
  "tags": ["tag1", "tag2"],
  "active": true,
  "age": 30
}
```
```

--------------------------------

### Serve Directories in Robyn

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/static-files.mdx

Use `app.serve_directory()` to serve static files. Configure for SPA builds with `index_file`, enable browsable listings with `show_files_listing=True`, or use default direct file serving.

```python
import os
from robyn import Robyn

app = Robyn(__file__)

# Mode 1: SPA / Build folder (e.g., React, Vue)
app.serve_directory(
    route="/",
    directory_path=os.path.join(os.path.dirname(__file__), "build"),
    index_file="index.html",
)

# Mode 2: Browsable file listing
app.serve_directory(
    route="/files",
    directory_path="./uploads",
    show_files_listing=True,
)

# Mode 3: Direct file serving (no listing, no index)
app.serve_directory(
    route="/assets",
    directory_path="./static",
)

app.start(port=8080)
```

--------------------------------

### Serve HTML and API Routes

Source: https://context7.com/sparckles/robyn/llms.txt

Demonstrates serving static HTML files, dynamic HTML strings, and standard JSON API endpoints.

```python
@app.get("/page")
async def serve_page(request: Request):
    return serve_html("./templates/index.html")

# Return HTML string directly
@app.get("/dynamic")
async def dynamic_html(request: Request):
    return html("<h1>Hello from Robyn</h1><p>Dynamic HTML content</p>")

# Combine static serving with API routes
@app.get("/api/health")
def health():
    return {"status": "ok"}

@app.get("/api/users")
async def get_users():
    return [{"id": 1, "name": "Alice"}]

app.start(port=8080)
```

--------------------------------

### Disable OpenAPI via CLI

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/openapi.mdx

Use this command to start the application without generating OpenAPI documentation.

```bash
python app.py --disable-openapi
```

--------------------------------

### Create index.html file

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/hosting.mdx

A simple HTML file to be served by the Robyn application.

```python
<h1> Hello World, this is Robyn framework! <h1>
```

--------------------------------

### Configure main.py for Railway deployment

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/hosting.mdx

The entry point script for the application. Ensure the host is set to 0.0.0.0 and the port uses the PORT environment variable.

```python
from robyn import Robyn, serve_html


app = Robyn(__file__)


@app.get("/hello")
async def h(request):
    print(request)
    return "Hello, world!"


@app.get("/")
async def get_page(request):
    return serve_html("./index.html")


if __name__ == "__main__":
    app.start(url="0.0.0.0", port=PORT)
```

--------------------------------

### Check Python Version

Source: https://github.com/sparckles/robyn/blob/main/README.md

Verify your installed Python version in the terminal. Robyn requires Python 3.10 or newer.

```bash
python --version
```

--------------------------------

### Run Server and Tests

Source: https://github.com/sparckles/robyn/blob/main/README.md

Commands to execute the server or run test suites.

```bash
poetry run test_server
```

```bash
pytest
```

```bash
pytest integration_tests
```

```bash
pytest unit_tests
```

--------------------------------

### GET /items/:id

Source: https://github.com/sparckles/robyn/blob/main/docs_src/public/llms.txt

Retrieves an item by its ID, supporting path parameters and query parameters with type coercion.

```APIDOC
## GET /items/:id

### Description
Retrieves an item specified by its ID. This endpoint demonstrates path parameters and query parameters with automatic type coercion.

### Method
GET

### Endpoint
/items/:id

### Parameters
#### Path Parameters
- **id** (int) - Required - The unique identifier of the item.

#### Query Parameters
- **q** (str) - Required - A query string parameter.
- **page** (int) - Optional - The page number, defaults to 1.

### Response
#### Success Response (200)
- **id** (int) - The ID of the item.
- **q** (str) - The query string parameter.
- **page** (int) - The page number.

### Response Example
```json
{
  "id": 1,
  "q": "example_query",
  "page": 1
}
```
```

--------------------------------

### Timeout Configuration via app.start()

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/timeout_configuration.mdx

Configure client and keep-alive timeouts directly within the `app.start()` method for fine-grained control.

```APIDOC
## POST /api/users

### Description
This endpoint allows for the creation of new user resources.

### Method
POST

### Endpoint
/api/users

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **username** (string) - Required - The desired username for the new user.
- **email** (string) - Required - The email address of the new user.
- **password** (string) - Required - The password for the new user.

### Request Example
```json
{
  "username": "johndoe",
  "email": "john.doe@example.com",
  "password": "securepassword123"
}
```

### Response
#### Success Response (201)
- **id** (string) - The unique identifier for the newly created user.
- **username** (string) - The username of the created user.
- **email** (string) - The email address of the created user.

#### Response Example
```json
{
  "id": "user-12345",
  "username": "johndoe",
  "email": "john.doe@example.com"
}
```
```

--------------------------------

### Configure Timeouts in app.start()

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/timeout_configuration.mdx

Set client and keep-alive timeouts directly within the application startup method.

```python
from robyn import Robyn

app = Robyn(__file__)

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

# Configure timeout settings
app.start(
    host="0.0.0.0", 
    port=8080,
    client_timeout=30,          # Client connection timeout (seconds)
    keep_alive_timeout=20       # Keep-alive timeout (seconds)
)
```

--------------------------------

### GET /sync/extra/*extra

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/request_object.mdx

Demonstrates how to capture dynamic, nested path segments using the *extra syntax in route definitions.

```APIDOC
## GET /sync/extra/*extra

### Description
Captures any additional segments in the URL path following the defined route prefix.

### Method
GET

### Endpoint
/sync/extra/*extra

### Parameters
#### Path Parameters
- **extra** (string) - Required - The captured path segments after /sync/extra/.

### Response
#### Success Response (200)
- **extra** (string) - The captured path segments as a single string.
```

--------------------------------

### Implement a RESTful API with Robyn

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/getting_started.mdx

Demonstrates handling various HTTP methods for a blog post resource using in-memory storage. Requires importing Robyn and Request from the robyn package.

```python
from robyn import Robyn, Request

app = Robyn(__file__)

# In-memory storage for demo
posts = {
    "1": {"id": "1", "title": "First Post", "content": "Hello World"},
    "2": {"id": "2", "title": "Second Post", "content": "Learning Robyn"}
}

# GET - Retrieve all posts
@app.get("/posts")
def get_posts(query_params):
    limit = int(query_params.get("limit", "10"))
    posts_list = list(posts.values())[:limit]
    return {"posts": posts_list, "total": len(posts)}

# GET - Retrieve specific post
@app.get("/posts/:id")
def get_post(path_params):
    post_id = path_params["id"]
    if post_id in posts:
        return {"post": posts[post_id]}
    return {"error": "Post not found"}, 404

# POST - Create new post
@app.post("/posts")
def create_post(request: Request):
    data = request.json()
    post_id = str(len(posts) + 1)
    new_post = {
        "id": post_id,
        "title": data.get("title", ""),
        "content": data.get("content", "")
    }
    posts[post_id] = new_post
    return {"message": "Post created", "post": new_post}, 201

# PUT - Update entire post
@app.put("/posts/:id")
def update_post(request: Request, path_params):
    post_id = path_params["id"]
    if post_id not in posts:
        return {"error": "Post not found"}, 404
    
    data = request.json()
    posts[post_id] = {
        "id": post_id,
        "title": data.get("title", ""),
        "content": data.get("content", "")
    }
    return {"message": "Post updated", "post": posts[post_id]}

# PATCH - Partial update
@app.patch("/posts/:id")
def patch_post(request: Request, path_params):
    post_id = path_params["id"]
    if post_id not in posts:
        return {"error": "Post not found"}, 404
    
    data = request.json()
    post = posts[post_id]
    
    # Update only provided fields
    if "title" in data:
        post["title"] = data["title"]
    if "content" in data:
        post["content"] = data["content"]
    
    return {"message": "Post updated", "post": post}

# DELETE - Remove post
@app.delete("/posts/:id")
def delete_post(path_params):
    post_id = path_params["id"]
    if post_id not in posts:
        return {"error": "Post not found"}, 404
    
    deleted_post = posts.pop(post_id)
    return {"message": "Post deleted", "post": deleted_post}
```

--------------------------------

### WebSocket API Methods and Properties

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/websockets.mdx

A reference guide for the methods and properties available on the websocket object within Robyn handlers.

```APIDOC
## WebSocket API Reference

### Description
The `websocket` object provides methods to interact with connected clients, including receiving messages, sending data, and managing the connection lifecycle.

### Methods
- **receive_text()**: Block until next message; raises `WebSocketDisconnect` on close.
- **receive_bytes()**: Block until next binary message; raises `WebSocketDisconnect` on close.
- **receive_json()**: Same as `receive_text()` but JSON-decoded.
- **send_text(data)**: Send string to this client.
- **send_bytes(data)**: Send binary data to this client.
- **send_json(data)**: Send JSON to this client.
- **broadcast(data)**: Send to all clients on this endpoint.
- **close()**: Close the connection server-side.

### Properties
- **id**: Connection UUID string.
- **query_params**: Query parameters from the connection URL.
```

--------------------------------

### Serve Static Files and Directories

Source: https://context7.com/sparckles/robyn/llms.txt

Configure directory serving for SPAs or static assets and serve individual files programmatically.

```python
from robyn import Robyn, Request, serve_file, serve_html, html
import os

app = Robyn(__file__)

# Serve a React/Vue SPA build directory
app.serve_directory(
    route="/",
    directory_path=os.path.join(os.path.dirname(__file__), "build"),
    index_file="index.html"
)
```

```python
# Serve static assets with browsable file listing
app.serve_directory(
    route="/files",
    directory_path="./uploads",
    show_files_listing=True
)
```

```python
# Serve assets without directory listing
app.serve_directory(
    route="/assets",
    directory_path="./static"
)
```

```python
# Serve individual files programmatically
@app.get("/download/report")
async def download_report(request: Request):
    return serve_file(
        "./reports/annual.pdf",
        file_name="annual-report-2024.pdf"  # Download filename
    )
```

--------------------------------

### Configure Timeouts via Environment Variables

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/timeout_configuration.mdx

Override default timeout settings using shell environment variables before starting the application.

```bash
# Set timeout configurations
export ROBYN_CLIENT_TIMEOUT=45
export ROBYN_KEEP_ALIVE_TIMEOUT=30

# Start your application
python app.py
```

--------------------------------

### Serve HTML files

Source: https://github.com/sparckles/robyn/blob/main/docs_src/src/pages/documentation/en/api_reference/file-uploads.mdx

Uses the serve_html utility to return an HTML file from the filesystem.

```python
from robyn import Robyn, Request, serve_html

app = Robyn(__file__)


@app.get("/")
async def h(request: Request):
    return serve_html("./index.html")

app.start(port=8080)
```