JWT Middleware with Authorization Headers

Complete AgentOS setup with JWT middleware for authentication and parameter injection using Authorization headers

AuthMiddleware validates the JWT in the Authorization header and injects its claims into endpoint parameters.

AuthMiddleware was named JWTMiddleware before v2.7. JWTMiddleware remains as an alias.

Code

jwt_middleware.py
from datetime import UTC, datetime, timedelta

import jwt
from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.models.openai import OpenAIResponses
from agno.os import AgentOS
from agno.run import RunContext
from agno.os.middleware.jwt import AuthMiddleware

# JWT Secret (use environment variable in production)
JWT_SECRET = "a-string-secret-at-least-256-bits-long"

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


# Define a tool that uses dependencies claims
def get_user_details(run_context: RunContext):
    """
    Get the current user's details.
    """
    dependencies = run_context.dependencies or {}
    return {
        "name": dependencies.get("name"),
        "email": dependencies.get("email"),
        "roles": dependencies.get("roles"),
    }


# Create agent
research_agent = Agent(
    id="user-agent",
    model=OpenAIResponses(id="gpt-5.2"),
    db=db,
    tools=[get_user_details],
    instructions="You are a user agent that can get user details if the user asks for them.",
)


agent_os = AgentOS(
    description="JWT Protected AgentOS",
    agents=[research_agent],
)

# Get the final app
app = agent_os.get_app()

# Add auth middleware to the app
# This middleware will automatically inject JWT values into request.state and is used in the relevant endpoints.
app.add_middleware(
    AuthMiddleware,
    verification_keys=[JWT_SECRET], # or use JWT_VERIFICATION_KEY environment variable
    algorithm="HS256",
    user_id_claim="sub",  # Extract user_id from 'sub' claim
    session_id_claim="session_id",  # Extract session_id from 'session_id' claim
    dependencies_claims=["name", "email", "roles"],
    validate=True,  # Verify the signature and expiration before using claims
)

if __name__ == "__main__":
    """
    Run your AgentOS with JWT parameter injection.

    Test by calling /agents/user-agent/runs with a message: "What do you know about me?"
    """
    # Test token with user_id and session_id:
    payload = {
        "sub": "user_123",  # This will be injected as user_id parameter
        "session_id": "demo_session_456",  # This will be injected as session_id parameter
        "exp": datetime.now(UTC) + timedelta(hours=24),
        "iat": datetime.now(UTC),
        # Dependency claims
        "name": "John Doe",
        "email": "john.doe@example.com",
        "roles": ["admin", "user"],
    }
    token = jwt.encode(payload, JWT_SECRET, algorithm="HS256")
    print("Test token:")
    print(token)
    agent_os.serve(app="jwt_middleware:app", port=7777, reload=True)

Usage

Set up your virtual environment

uv venv --python 3.12
source .venv/bin/activate

Set Environment Variables

export OPENAI_API_KEY=your_openai_api_key

Install dependencies

uv pip install -U agno openai pyjwt "fastapi[standard]" uvicorn sqlalchemy pgvector "psycopg[binary]"

Setup PostgreSQL Database

# Using Docker
docker run -d \
  --name agno-postgres \
  -e POSTGRES_DB=ai \
  -e POSTGRES_USER=ai \
  -e POSTGRES_PASSWORD=ai \
  -p 5532:5432 \
  pgvector/pgvector:pg17

Run Example

python jwt_middleware.py

The server will start and print a test JWT token to the console.

Test JWT Authentication

Test with the generated token:

# Use the token printed in the console
export TOKEN="eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9..."

curl --location 'http://localhost:7777/agents/user-agent/runs' \
    --header 'Content-Type: application/x-www-form-urlencoded' \
    --header "Authorization: Bearer $TOKEN" \
    --data-urlencode 'message=What do you know about me?'

Test without a token:

curl --location 'http://localhost:7777/agents/user-agent/runs' \
    --header 'Content-Type: application/x-www-form-urlencoded' \
    --data-urlencode 'message=What do you know about me?'

This returns 401 with {"detail": "Authorization header missing"}. Tokens with an invalid signature or expired exp claim are also rejected.

Check the AgentOS API docs: Visit http://localhost:7777/docs to see all available endpoints.

How It Works

  1. JWT Generation: The example creates a test JWT token with user claims
  2. Middleware Setup: JWT middleware extracts claims from the Authorization: Bearer <token> header
  3. Parameter Injection: The middleware automatically injects:
    • user_id from the sub claim
    • session_id from the session_id claim
    • dependencies dict with name, email, and roles
  4. Agent Tools: RunContext.dependencies gives tools the verified claims without exposing them as model-supplied tool arguments

Next Steps