> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agno.com/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
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"}
)
```
Error responses must be returned as `JSONResponse` objects to ensure proper serialization and HTTP status codes.
## Adding Middleware to AgentOS
```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()
```
```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)
```
```python theme={null}
if __name__ == "__main__":
agent_os.serve(app="custom_middleware:app", reload=True)
```
## Developer Resources
Rate limiting and request logging middleware implementation.
Custom FastAPI app with JWT middleware and AgentOS integration.
Official FastAPI middleware documentation and examples.