Custom Middleware
Create custom middleware for rate limiting, logging, security, and monitoring in AgentOS
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.
The rate limiter below is an in-memory, single-process example. Each worker has its own counters, and request.client.host may identify a proxy. Use a shared limiter and a trusted client-identity policy for a multi-worker deployment.
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("Body preview: %s", body[:2048].decode("utf-8", errors="replace"))
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 responseError 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"}
)Return a Starlette Response with the intended status and content type. JSONResponse is a convenient option for JSON errors.
Adding Middleware to AgentOS
Use a running PostgreSQL database with the ai user, password, and database on port 5532, or replace the example db_url. Install the server and driver dependencies and set the model key in its terminal:
uv pip install -U "agno[os]" openai "psycopg[binary]"
export OPENAI_API_KEY="your_openai_api_key"Assemble the RateLimitMiddleware, LoggingMiddleware, and SecurityHeadersMiddleware definitions above in custom_middleware.py, then append these steps in order. Start it with python custom_middleware.py. Keep body logging disabled for content that should not enter application logs.
Create AgentOS App
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)