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)

""" Log all requests with timing and metadata """
import logging
import time
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

class LoggingMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, log_body: bool = False):
        super().__init__(app)
        self.log_body = log_body
        self.logger = logging.getLogger("request_logger")

    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        client_ip = request.client.host if request.client else "unknown"
        
        # Log request
        self.logger.info(f"Request: {request.method} {request.url.path} from {client_ip}")
        
        # Optionally log body
        if self.log_body and request.method in ["POST", "PUT", "PATCH"]:
            body = await request.body()
            if body:
                self.logger.info(f"Body: {body.decode()}")
        
        response = await call_next(request)
        
        # Log response
        duration = (time.time() - start_time) * 1000
        self.logger.info(f"Response: {response.status_code} in {duration:.1f}ms")
        
        return response

  """ Add security headers to all responses """
  from fastapi import Request
  from starlette.middleware.base import BaseHTTPMiddleware

  class SecurityHeadersMiddleware(BaseHTTPMiddleware):
      def __init__(self, app):
          super().__init__(app)

      async def dispatch(self, request: Request, call_next):
          response = await call_next(request)
          
          # Add security headers
          response.headers["X-Content-Type-Options"] = "nosniff"
          response.headers["X-Frame-Options"] = "DENY"
          response.headers["X-XSS-Protection"] = "1; mode=block"
          response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
          response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
          
          return response

  """ Add unique request IDs for tracing """
  import uuid
  from fastapi import Request
  from starlette.middleware.base import BaseHTTPMiddleware

  class RequestIDMiddleware(BaseHTTPMiddleware):
      def __init__(self, app):
          super().__init__(app)

      async def dispatch(self, request: Request, call_next):
          # Generate unique request ID
          request_id = str(uuid.uuid4())
          
          # Store in request state
          request.state.request_id = request_id
          
          # Process request
          response = await call_next(request)
          
          # Add to response headers
          response.headers["X-Request-ID"] = request_id
          
          return response

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 OpenAIResponses

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

agent = Agent(
    name="Basic Agent",
    model=OpenAIResponses(id="gpt-5.2"),
    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