Custom Middleware

Each middleware wraps your application to intercept requests and responses, enabling you to implement cross-cutting concerns like authentication, logging, and rate limiting. AgentOS supports any FastAPI/Starlette middleware. You can create custom middleware for logging, rate limiting, monitoring, security, and more.

Creating Custom Middleware

Middleware in AgentOS follows the FastAPI/Starlette pattern using BaseHTTPMiddleware. See the following common middleware examples:

""" Rate limiting middleware that limits requests per IP address """
import time
from collections import defaultdict, deque
from fastapi import Request
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)

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"}
            )

Error responses must be returned as JSONResponse objects to ensure proper serialization and HTTP status codes.

Adding Middleware to AgentOS

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()

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)

Serve your AgentOS

if __name__ == "__main__":
    agent_os.serve(app="custom_middleware:app", reload=True)

Developer Resources