Gemini Agent + FastAPI: The Production-Ready Boilerplate for AI Chat APIs

Most Python web developers reach for FastAPI when they need to expose an HTTP API. Most Genkit Python examples show flows running from the command line. PR #5687 closes that gap with a working boilerplate for wiring a Gemini agent into a FastAPI app, anchored by the genkit_fastapi_handler decorator. This post walks through that boilerplate, explains what the decorator actually does, and covers the parts that matter in production: streaming, auth, CORS, and session handling.

What the decorator does

genkit_fastapi_handler wraps a Genkit flow or a function that returns one, and converts it into a FastAPI request handler. The wrapped function becomes a standard FastAPI endpoint. You decorate in this order: FastAPI route on the outside, then the Genkit handler, then the flow.

from fastapi import FastAPI
from genkit import Genkit
from genkit.plugins.fastapi import genkit_fastapi_handler
from genkit.plugins.google_genai import GoogleAI

ai = Genkit(plugins=[GoogleAI()], model='googleai/gemini-flash-latest')
app = FastAPI()

@app.post('/chat', response_model=None)
@genkit_fastapi_handler(ai)
@ai.flow()
async def chat_flow(prompt: str) -> str:
    response = await ai.generate(prompt=prompt)
    return response.text

The handler expects the request body to be wrapped in {"data": ...}. A request to /chat looks like:

curl -X POST http://localhost:8000/chat \
  -H 'Content-Type: application/json' \
  -d '{"data": "What is the capital of France?"}'

The response is {"result": "Paris."}. That envelope format is Genkit’s standard callable protocol, shared with the Firebase Dev UI and the TypeScript SDK. If the body lacks data, the handler returns a 400 error rather than passing None to your flow.

Looking at the handler source directly, the key dispatch is in handler():

body = await request.json()
if 'data' not in body:
    err = GenkitError(status='INVALID_ARGUMENT', message='Action request must be wrapped in {"data": ...} object')
    return Response(status_code=400, ...)

# ... auth, streaming check ...

response = await flow.run(body.get('data'), context=action_context)
return {'result': _to_dict(response.response)}

response_model=None on the FastAPI route is required. Without it, FastAPI tries to validate the return against a Pydantic schema and fails because the handler returns either a dict or a StreamingResponse depending on the client’s Accept header.

Full boilerplate: chat agent with sessions

A chat agent needs to maintain conversation history across requests. The standard approach is to pass a session_id in each request and store messages server-side. Here is a working boilerplate:

import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel

from genkit import Genkit
from genkit.plugins.fastapi import genkit_fastapi_handler
from genkit.plugins.google_genai import GoogleAI

ai = Genkit(plugins=[GoogleAI()], model='googleai/gemini-2.0-flash')
app = FastAPI(title='Chat API')

app.add_middleware(
    CORSMiddleware,
    allow_origins=['http://localhost:3000'],
    allow_methods=['POST'],
    allow_headers=['*'],
)


class ChatInput(BaseModel):
    message: str
    session_id: str | None = None


class ChatOutput(BaseModel):
    reply: str
    session_id: str


@app.post('/chat', response_model=None)
@genkit_fastapi_handler(ai)
@ai.flow()
async def chat_flow(input: ChatInput) -> ChatOutput:
    session_id = input.session_id or 'default'
    response = await ai.generate(
        system='You are a helpful assistant.',
        prompt=input.message,
    )
    return ChatOutput(reply=response.text, session_id=session_id)


if __name__ == '__main__':
    uvicorn.run(app, host='0.0.0.0', port=8000)

Run it with genkit start -- uvicorn main:app --reload to get the Dev UI at localhost:4000, or plain uvicorn main:app for production. The Dev UI reflection server starts automatically in a background thread when GENKIT_ENV=dev is set; you do not need to wire any lifespan handlers.

Streaming: how SSE chunks flow through the handler

The handler detects whether the client wants streaming by checking the Accept header. If it contains text/event-stream, the handler wraps the flow in a StreamingResponse:

accept = request.headers.get('accept', '')
stream = 'text/event-stream' in accept or request.query_params.get('stream') == 'true'

if stream:
    async def event_stream() -> AsyncIterator[str]:
        stream_response = flow.stream(body.get('data'), context=action_context)
        async for chunk in stream_response.stream:
            yield f'data: {json.dumps({"message": _to_dict(chunk)})}\n\n'
        result = await stream_response.response
        yield f'data: {json.dumps({"result": _to_dict(result)})}\n\n'

    return StreamingResponse(event_stream(), media_type='text/event-stream')

Each chunk arrives as data: {"message": ...}\n\n. The final message is data: {"result": ...}\n\n containing the complete response. The ?stream=true query param is a fallback for clients that cannot set the Accept header (certain mobile SDK versions, some proxy configurations).

On the client side, consuming the stream in JavaScript:

const response = await fetch('/chat', {
    method: 'POST',
    headers: {
        'Content-Type': 'application/json',
        'Accept': 'text/event-stream',
    },
    body: JSON.stringify({ data: { message: 'Hello!', session_id: 'user123' } }),
});

const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const text = decoder.decode(value);
    const lines = text.split('\n').filter(l => l.startsWith('data: '));
    for (const line of lines) {
        const payload = JSON.parse(line.slice(6));
        if (payload.message) process.stdout.write(payload.message);
        if (payload.result) console.log('\nDone:', payload.result);
    }
}

Auth patterns

The handler accepts a context_provider argument: a callable that receives the FastAPI Request wrapped in a RequestData object and returns a dict. That dict becomes the context passed to the flow’s action run context.

from fastapi import HTTPException
from genkit.plugin_api import RequestData

def verify_api_key(request_data: RequestData) -> dict:
    api_key = request_data.headers.get('x-api-key', '')
    if api_key != 'expected-key':
        raise HTTPException(status_code=401, detail='Invalid API key')
    return {'user_id': 'authenticated_user'}

@app.post('/chat', response_model=None)
@genkit_fastapi_handler(ai, context_provider=verify_api_key)
@ai.flow()
async def chat_flow(input: ChatInput) -> ChatOutput:
    # context is available via ActionRunContext in the flow's internals
    response = await ai.generate(prompt=input.message)
    return ChatOutput(reply=response.text, session_id='default')

The context_provider can also be async. If it raises an HTTPException, FastAPI handles it before the flow runs. This is the recommended place to put bearer token validation, session lookup, or rate-limit checks.

For more complex auth (JWT validation, database lookups, Firestore session reads), async context providers let you await those operations without blocking the event loop:

async def load_user_context(request_data: RequestData) -> dict:
    token = request_data.headers.get('authorization', '').removeprefix('Bearer ')
    user = await verify_jwt(token)  # your async JWT library
    if not user:
        raise HTTPException(status_code=401, detail='Unauthorized')
    return {'user_id': user.id, 'plan': user.plan}

CORS setup for browser clients

Standard FastAPI CORSMiddleware applies. The key detail is that if you’re using streaming, the Allow-Headers must include Accept so the browser can send text/event-stream:

app.add_middleware(
    CORSMiddleware,
    allow_origins=['https://your-frontend.com'],
    allow_credentials=True,
    allow_methods=['POST', 'OPTIONS'],
    allow_headers=['Content-Type', 'Accept', 'Authorization', 'X-Api-Key'],
    expose_headers=['Content-Type'],
)

In development, allow_origins=['*'] is fine. In production, pin it to your actual origin. If you forget to expose Content-Type in expose_headers, some browsers will refuse to read the SSE stream even after the request succeeds.

Contrast with serve_agent

PR #5686 adds serve_agent() and serve_flow() as one-liner deployment functions. They create a full FastAPI app for you:

# serve_agent/serve_flow style (article #37) — simpler, less control
from genkit.plugins.fastapi import serve_agent
app = serve_agent(my_agent)

The genkit_fastapi_handler decorator is more work to set up but gives you full control over the FastAPI app. You decide the URL structure, you add your own middleware, you compose multiple flows into one app, and you can mix Genkit endpoints with non-Genkit routes in the same server.

A concrete case where the decorator wins: an existing FastAPI app that already has auth middleware, Prometheus metrics, and a /health endpoint. Dropping genkit_fastapi_handler onto a new route means your new AI endpoint participates in all of that with zero additional wiring:

# existing FastAPI app with auth, metrics, health check
# add an AI endpoint alongside regular routes

@app.post('/summarize', response_model=None)
@genkit_fastapi_handler(ai)
@ai.flow()
async def summarize(text: str) -> str:
    response = await ai.generate(
        system='Summarize this text in one paragraph.',
        prompt=text,
    )
    return response.text

The serve_agent() approach works when you’re deploying a standalone agent with no existing web infrastructure. The decorator approach works when Genkit is one feature inside a larger application.

Session ID handling across requests

The Genkit genkit_fastapi_handler is stateless by default: each request starts a fresh flow execution. If your agent needs conversation history across requests, you have two choices:

Client-managed history: the client sends the full message list with each request. The flow reads messages from the input and passes them to ai.generate(). This is simple but means the client holds the state.

class ConversationInput(BaseModel):
    user_message: str
    history: list[dict] = []

@app.post('/chat', response_model=None)
@genkit_fastapi_handler(ai)
@ai.flow()
async def chat_with_history(input: ConversationInput) -> dict:
    from genkit.model import Message
    history = [Message.model_validate(m) for m in input.history]
    response = await ai.generate(
        messages=history,
        prompt=input.user_message,
    )
    return {
        'reply': response.text,
        'history': [m.model_dump() for m in response.messages],
    }

Server-managed sessions: the context provider loads session state from a store (Firestore, Redis), and the flow reads and writes it. This is more complex but keeps the client payload small. The FirestoreSessionStore (article #36, pending merge) is built specifically for this pattern.

For prototyping, client-managed history is fine. For production where clients are mobile apps or browsers that lose state on page reload, server-managed sessions are worth the complexity.

The fastapi-bugbot sample

The best reference for a complete working app is py/samples/fastapi-bugbot/src/main.py. It shows three flows (analyze_security, analyze_bugs, analyze_style) running in parallel via asyncio.gather(), a review_code flow combining them, and both direct FastAPI endpoints and genkit_fastapi_handler wrapped endpoints on the same app:

# Direct FastAPI endpoint — no Genkit protocol envelope
@app.post('/review')
async def review(input: CodeInput) -> Analysis:
    return await review_code(input)

# Genkit handler endpoint — requires {"data": ...} envelope, enables Dev UI inspection
@app.post('/flow/review', response_model=None)
@genkit_fastapi_handler(ai)
def flow_review() -> Flow[CodeInput, Analysis, Never]:
    return review_code

Both call the same underlying review_code flow. The direct endpoint is easier to call from code that doesn’t know the Genkit protocol. The handler endpoint shows up in the Dev UI at localhost:4000 during development, where you can trigger it, inspect the trace, and replay it with different inputs.