FastAPI Integration

Use ai-query with FastAPI for custom HTTP endpoints

Integrate ai-query with FastAPI when you need custom HTTP endpoints, middleware, or want to use your own server infrastructure instead of the built-in serve().

Quick Integration (AgentRouter)

The easiest way to serve an agent is using the AgentRouter. This adds a standard set of endpoints (/chat, /invoke, /state) to your application.

from fastapi import FastAPI
from ai_query.agents import Agent, SQLiteStorage
from ai_query.adapters.fastapi import AgentRouter
from ai_query.providers import openai

app = FastAPI()

# 1. Define Agent
agent = Agent(
    "support-bot",
    model=openai("gpt-4o"),
    storage=SQLiteStorage("agents.db")
)

# 2. Mount Router
# This creates routes like POST /api/agent/chat, POST /api/agent/invoke
app.include_router(AgentRouter(agent), prefix="/api/agent")

Basic Chat Endpoint (Manual)

If you only need a simple chat endpoint and don’t want the full agent API:

from fastapi import FastAPI
from pydantic import BaseModel
from ai_query import generate_text
from ai_query.providers import openai

app = FastAPI()

class ChatRequest(BaseModel):
    message: str

class ChatResponse(BaseModel):
    response: str

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    result = await generate_text(
        model=openai("gpt-4o"),
        messages=[{"role": "user", "content": request.message}]
    )
    return ChatResponse(response=result.text)

Streaming with SSE

Use StreamingResponse for real-time streaming.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from ai_query import stream_text
from ai_query.providers import openai

app = FastAPI()

class ChatRequest(BaseModel):
    message: str

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        result = stream_text(
            model=openai("gpt-4o"),
            messages=[{"role": "user", "content": request.message}]
        )
        async for chunk in result.text_stream:
            # SSE format
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

Advanced: Custom Agent Integration

If AgentRouter doesn’t fit your needs, you can integrate the Agent class manually.

from fastapi import FastAPI, Depends, HTTPException
from ai_query.agents import Agent, SQLiteStorage
from ai_query.providers import openai

app = FastAPI()

agent = Agent("my-agent", model=openai("gpt-4o"))

@app.on_event("startup")
async def startup():
    await agent.start()

@app.post("/custom-chat")
async def chat(message: str):
    # Use handle_request for consistent behavior
    result = await agent.handle_request({
        "action": "chat",
        "message": message
    })
    return result