> ## Documentation Index > Fetch the complete documentation index at: https://agno-v2.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. # Custom Middleware > Create custom middleware for rate limiting, logging, security, and monitoring in AgentOS v2.1.0 (Introduced in v2.1.0) 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](https://fastapi.tiangolo.com/tutorial/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: ```python Rate Limiting theme={null} """ 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) ``` ```python Request Logging theme={null} """ 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 ``` ```python Security Headers theme={null} """ 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 ``` ```python Request ID theme={null} """ 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: ```python theme={null} 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"} ) ``` > [!NOTE] > > Error responses must be returned as `JSONResponse` objects to ensure proper serialization and HTTP status codes. ## Adding Middleware to AgentOS **1. Create AgentOS App** ```python custom_middleware.py theme={null} 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() ``` **2. Add Custom Middleware** ```python theme={null} # 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** ```python theme={null} if __name__ == "__main__": agent_os.serve(app="custom_middleware:app", reload=True) ``` ## Developer Resources > **[Custom Middleware](https://agno-v2.mintlify.site/agent-os/usage/middleware/custom-middleware.md)** > > Rate limiting and request logging middleware implementation. > **[Custom FastAPI + JWT](https://agno-v2.mintlify.site/agent-os/usage/middleware/custom-fastapi-jwt.md)** > > Custom FastAPI app with JWT middleware and AgentOS integration. > **[FastAPI Middleware](https://fastapi.tiangolo.com/tutorial/middleware/)** > > Official FastAPI middleware documentation and examples. This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.