Skip to main content
AgentOS supports any FastAPI/Starlette middleware. You can create custom middleware for logging, rate limiting, monitoring, security, and more.

​
Creating Custom Middleware

Custom middleware in AgentOS follows the FastAPI/Starlette middleware pattern using BaseHTTPMiddleware: 
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import Request, Response

class CustomMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, param1: str = "default"):
        super().__init__(app)
        self.param1 = param1

    async def dispatch(self, request: Request, call_next) -> Response:
        # Before request processing
        start_time = time.time()
        
        # Process request
        response = await call_next(request)
        
        # After request processing  
        process_time = time.time() - start_time
        response.headers["X-Process-Time"] = str(process_time)
        
        return response

​
Common Use Cases

These are some examples demonstrating how to add some common custom middlewares: 
""" Rate limiting middleware that limits requests per IP address """
import time
from collections import defaultdict, deque
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware

class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, requests_per_minute: int = 60):
        super().__init__(app)
        self.requests_per_minute = requests_per_minute
        self.request_history = defaultdict(lambda: deque())

    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host if request.client else "unknown"
        current_time = time.time()
        
        # Clean old requests
        history = self.request_history[client_ip]
        while history and current_time - history[0] > 60:
            history.popleft()
        
        # Check rate limit
        if len(history) >= self.requests_per_minute:
            return JSONResponse(
                status_code=429,
                content={"detail": "Rate limit exceeded"}
            )
        
        history.append(current_time)
        return await call_next(request)

​
Adding Middleware to AgentOS

1

Create AgentOS App

custom_middleware.py
from agno.os import AgentOS
from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.models.openai import OpenAIChat

db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")

agent = Agent(
    name="Basic Agent",
    model=OpenAIChat(id="gpt-5-mini"),
    db=db,
)

agent_os = AgentOS(agents=[agent])
app = agent_os.get_app()
2

Add Custom Middleware

# Add your custom middleware
app.add_middleware(
    RateLimitMiddleware,
    requests_per_minute=100
)

app.add_middleware(
    LoggingMiddleware,
    log_body=False
)

app.add_middleware(SecurityHeadersMiddleware)
3

Serve your AgentOS

if __name__ == "__main__":
    agent_os.serve(app="custom_middleware:app", reload=True)
See the following full examples:

​
Error Handling

Handle exceptions in middleware: 
from fastapi import Request
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware

import logging

logger = logging.getLogger(__name__)

class ErrorHandlingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        try:
            response = await call_next(request)
            return response
        except Exception as e:
            # Log the error
            logger.error(f"Request failed: {e}")
            
            # Return error response as JSONResponse
            return JSONResponse(
                status_code=500,
                content={"detail": "Internal server error"}
            )
It is important to note that the error response should be a JSONResponse.

​
Developer Resources

⌘I