> ## Documentation Index
> Fetch the complete documentation index at: https://spacesail.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 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`:

```python theme={null}
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:

<CodeGroup>
  ```python Rate Limiting theme={null}
  """ 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)
  ```

  ```python Request Logging theme={null}
  """ Log all requests with timing and metadata """
  import logging
  import time
  from fastapi import Request
  from fastapi.responses import JSONResponse
  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
  ```
</CodeGroup>

## Adding Middleware to AgentOS

<Steps>
  <Step title="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 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()
    ```
  </Step>

  <Step title="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)
    ```
  </Step>

  <Step title="Serve your AgentOS">
    ```python theme={null}
    if __name__ == "__main__":
        agent_os.serve(app="custom_middleware:app", reload=True)
    ```
  </Step>
</Steps>

See the following full examples:

* [Custom Middleware](/examples/agent-os/middleware/custom-middleware)
* [Custom FastAPI + JWT](/examples/agent-os/middleware/custom-fastapi-jwt)

## 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"}
            )
```

It is important to note that the error response should be a `JSONResponse`.

## Developer Resources

* [Custom Middleware Example](/examples/agent-os/middleware/custom-middleware)
* [FastAPI Middleware Documentation](https://fastapi.tiangolo.com/tutorial/middleware/)