Custom Middleware

Create and add your own middleware to AgentOS applications

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

""" 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:

• Custom Middleware
• Custom FastAPI + JWT

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

• Custom Middleware Example
• FastAPI Middleware Documentation