from datetime import datetime, timedelta, UTC
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.os.middleware import JWTMiddleware
from agno.tools.hackernews import HackerNewsTools
from fastapi import FastAPI, Form, HTTPException
# 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")
# Create agent
research_agent = Agent(
id="research-agent",
name="Research Agent",
model=OpenAIResponses(id="gpt-5.2"),
db=db,
tools=[HackerNewsTools()],
add_history_to_context=True,
markdown=True,
)
# Create custom FastAPI app
app = FastAPI(
title="Example Custom App",
version="1.0.0",
)
# Add Agno JWT middleware to your custom FastAPI app
app.add_middleware(
JWTMiddleware,
verification_keys=[JWT_SECRET],
excluded_route_paths=[
"/auth/login"
], # We don't want to validate the token for the login endpoint
validate=True, # Set validate to False to skip token validation
)
# Custom routes that use JWT
@app.post("/auth/login")
async def login(username: str = Form(...), password: str = Form(...)):
"""Login endpoint that returns JWT token"""
if username == "demo" and password == "password":
payload = {
"sub": "user_123",
"username": username,
"exp": datetime.now(UTC) + timedelta(hours=24),
"iat": datetime.now(UTC),
}
token = jwt.encode(payload, JWT_SECRET, algorithm="HS256")
return {"access_token": token, "token_type": "bearer"}
raise HTTPException(status_code=401, detail="Invalid credentials")
# Clean AgentOS setup with tuple middleware pattern! ✨
agent_os = AgentOS(
description="JWT Protected AgentOS",
agents=[research_agent],
base_app=app,
)
# Get the final app
app = agent_os.get_app()
if __name__ == "__main__":
"""
Run your AgentOS with JWT middleware applied to the entire app.
Test endpoints:
1. POST /auth/login - Login to get JWT token
2. GET /config - Protected route (requires JWT)
"""
agent_os.serve(
app="custom_fastapi_jwt:app", port=7777, reload=True
)