Skip to content

FastAPI tutorial

In this tutorial, you’ll build Bargain Chef, a standalone Genkit backend on FastAPI that exposes a recipe-generating flow over HTTP. It uses two AI patterns Genkit simplifies: streaming structured output and tool calling.

What you’ll build

Section titled “What you’ll build”

For each request, your server prompts Gemini to draft a recipe, and the model calls a tool to look up mock grocery sale prices so it can prefer on-sale ingredients. The server streams the recipe back field-by-field as it’s generated, so clients see progress before the full recipe is ready.

You can find the finished code on GitHub.

Prerequisites

Section titled “Prerequisites”
  • Python 3.10 or later
  • uv package manager

This tutorial assumes you’re already familiar with building FastAPI applications.

Set up the application

Section titled “Set up the application”

Create the FastAPI project

Section titled “Create the FastAPI project”
Terminal window
mkdir bargain-chef
cd bargain-chef
uv init --no-readme --python 3.10

Install packages

Section titled “Install packages”

Install the Genkit CLI:

Terminal window
curl -sL cli.genkit.dev | bash

Then install the packages you need in your project:

Terminal window
uv add fastapi uvicorn genkit genkit-google-genai genkit-fastapi

These packages include:

  • fastapi: The async Python web framework that serves your endpoint.
  • uvicorn: The ASGI server that runs your FastAPI app.
  • genkit: Core Genkit SDK.
  • genkit-google-genai: Plugin that connects Genkit to Google’s Gemini models.
  • genkit-fastapi: Serves Genkit flows as FastAPI routes, with SSE streaming.

Configure a model API key

Section titled “Configure a model API key”

This tutorial uses the Gemini API from Google AI Studio. Get a key from Google AI Studio, then set the GEMINI_API_KEY environment variable to your key:

Terminal window
export GEMINI_API_KEY=<your API key>

Optional: install the Genkit agent skills

Section titled “Optional: install the Genkit agent skills”

If you’re coding with an AI assistant, install the Genkit Agent Skills so it has structured guidance on Genkit APIs, patterns, and common errors:

Terminal window
npx skills add genkit-ai/skills

See Develop with AI for tool-specific installation instructions.

Create the backend

Section titled “Create the backend”

The backend handles requests from clients. For each request, it prompts Gemini to draft a recipe, lets the model call a tool to look up today’s grocery sale prices, and streams the partial recipe back to the caller as it’s generated.

The whole pipeline is a single Genkit flow. A flow is a special Genkit function with built-in observability, type safety, and tooling integration.

You’ll build the backend in four parts:

  1. Initialize Genkit and register Gemini as the model provider.
  2. Define a tool the model can call to fetch sale prices.
  3. Describe the recipe shape with Pydantic so Genkit can validate the final output and stream partial recipe chunks.
  4. Define the flow that ties everything together.

Replace the contents of main.py with the following:

main.py
from datetime import datetime
from typing import Literal


from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field


from genkit import FlowContext, Genkit
from genkit_fastapi import serve_flow
from genkit_google_genai import GeminiConfig, GoogleAI


ai = Genkit(
    plugins=[GoogleAI()],
    model=GoogleAI.gemini('gemini-flash-latest'),
)


class SaleItem(BaseModel):
    name: str
    price: str


class GetIngredientsInput(BaseModel):
    day_type: Literal['weekday', 'weekend'] = Field(
        description='Whether to fetch weekday or weekend sale prices.',
    )


@ai.tool()
async def get_ingredients_on_sale(input: GetIngredientsInput) -> list[SaleItem]:
    """Returns the ingredients on sale at the local grocery store, with prices.


    The sale set differs between weekdays and weekends.
    """
    # Mock data: in a real app, query a pricing database.
    if input.day_type == 'weekend':
        return [
            SaleItem(name='chicken breast', price='$2.99/lb'),
            SaleItem(name='pasta', price='$0.79'),
            SaleItem(name='canned tomatoes', price='$0.99'),
            SaleItem(name='garlic', price='$0.50 / head'),
            SaleItem(name='olive oil', price='$6.99'),
        ]
    return [
        SaleItem(name='eggs', price='$3.49 / dozen'),
        SaleItem(name='spinach', price='$1.99'),
        SaleItem(name='parmesan', price='$4.99'),
        SaleItem(name='lemons', price='$0.50 each'),
        SaleItem(name='rice', price='$2.49'),
        SaleItem(name='butter', price='$3.99'),
    ]


class RecipeIngredient(BaseModel):
    name: str
    quantity: str
    on_sale: bool = Field(serialization_alias='onSale')


class Recipe(BaseModel):
    title: str
    description: str
    servings: int
    ingredients: list[RecipeIngredient]
    steps: list[str]


class BargainChefInput(BaseModel):
    craving: str = Field(description='What the user feels like eating right now.')


@ai.flow(name='bargainChefFlow')
async def bargain_chef_flow(
    input: BargainChefInput,
    ctx: FlowContext[Recipe],
) -> Recipe:
    today = datetime.now().strftime('%A')


    stream_response = ai.generate_stream(
        prompt=(
            f'Today is {today}. The user is craving: {input.craving}.\n\n'
            'Call the get_ingredients_on_sale tool with the day_type that matches today. '
            'Saturday and Sunday are weekends; all other days are weekdays. '
            'Then propose ONE recipe that takes advantage of those deals. For each '
            "ingredient, set on_sale=true if it appears in the tool's response, "
            'false otherwise.'
        ),
        tools=[get_ingredients_on_sale],
        output_schema=Recipe,
        config=GeminiConfig(temperature=0.7, thinking_level='LOW'),
    )


    async for chunk in stream_response.stream:
        if chunk.output:
            ctx.send_chunk(chunk.output)


    response = await stream_response.response
    if response.output is None:
        raise RuntimeError(f'Recipe generation failed: {response.error}')
    return response.output


app = FastAPI()


# allow_origins=['*'] lets any browser frontend call the endpoint during
# development. Before deploying, restrict it to the origins you actually serve.
app.add_middleware(
    CORSMiddleware,
    allow_origins=['*'],
    allow_methods=['*'],
    allow_headers=['*'],
)


# POST /bargainChefFlow — path defaults to the flow's name
app.include_router(serve_flow(bargain_chef_flow))

A few details are worth noting before you run the backend:

  • Final output and streamed chunks: output_schema=Recipe validates the complete recipe returned at the end and parses partial Recipe objects as JSON tokens arrive. Annotating ctx: FlowContext[Recipe] types the streamed chunks sent via ctx.send_chunk. Return response.output (or response.text), never a raw ModelResponse, whose request field carries your system prompt, tools, retrieved documents, and history.
  • Shared Python types: The Pydantic models (Recipe, RecipeIngredient, BargainChefInput) define the request and response shapes once. serialization_alias='onSale' puts camelCase onSale on the JSON and SSE wire while keeping snake_case in Python.
  • The get_ingredients_on_sale tool: The model decides when to call it based on the prompt, and the typed GetIngredientsInput forces the model to pass day_type='weekday' or 'weekend'. In a real app, the tool would query a pricing database, inventory system, or third-party API.
  • ctx.send_chunk: Each call pushes the latest partial recipe to the client, giving it a typed view of the generated JSON as it grows. After the stream completes, the flow awaits stream_response.response so the HTTP request resolves with a validated recipe.
  • Failed generation: generate_stream doesn’t raise when the model fails (for example, an overloaded model). The response has no output and carries the failure on response.error. The flow raises with that failure, so the client gets a generic Internal Error (a 500, or a final error event on a streaming request) and your server log records the status and message. See Error types.

Expose the flow over HTTP

Section titled “Expose the flow over HTTP”

serve_flow returns a FastAPI APIRouter you mount with app.include_router. By default the route is POST /<flow name> (here /bargainChefFlow). When the client sends Accept: text/event-stream or ?stream=true, the endpoint streams partial chunks as Server-Sent Events; otherwise it returns the final recipe as JSON. Pass base_path= or a router prefix= if you want a different URL.

Attach request context

Section titled “Attach request context”

Use context_dependency on serve_flow to wire FastAPI Depends into Genkit. FastAPI resolves the dependency before the Genkit handler runs, and the returned dict becomes ctx.context in the flow and its tools. Replace the app.include_router(serve_flow(bargain_chef_flow)) line in main.py with:

from fastapi import Header, HTTPException


async def user_context(authorization: str | None = Header(default=None)) -> dict[str, object]:
    token = (authorization or '').removeprefix('Bearer ').strip()
    if not token:
        raise HTTPException(status_code=401, detail='Missing bearer token')
    return {'uid': token}


app.include_router(
    serve_flow(bargain_chef_flow, context_dependency=user_context),
)

Reject a request from a dependency with HTTPException, as above. Don’t raise PublicError there: FastAPI runs dependencies before Genkit sees the request, so it becomes a plain-text 500. See Errors and timeouts for other failures.

To read context from the raw request instead of Depends, pass a context_provider to @genkit_fastapi_handler:

from genkit import FlowContext, PublicError
from genkit.web import RequestData
from genkit_fastapi import genkit_fastapi_handler


async def auth_provider(req: RequestData) -> dict[str, object]:
    token = (req.headers.get('authorization') or '').removeprefix('Bearer ').strip()
    if not token:
        raise PublicError('UNAUTHENTICATED', 'Missing bearer token')
    return {'uid': token}


@app.post('/custom-chef')
@genkit_fastapi_handler(ai, context_provider=auth_provider)
@ai.flow()
async def custom_chef(prompt: str, ctx: FlowContext) -> str:
    uid = ctx.context.get('uid')
    res = await ai.generate(prompt=f'Answer for {uid}: {prompt}')
    if res.error:
        raise RuntimeError(f'Answer failed: {res.error}')
    return res.text

When FastAPI owns the route signature (for custom Depends(...), rate limiters, or typed query parameters), delegate to handle_genkit_request for full JSON + SSE handling, or stream a handle directly with genkit.web.to_sse:

from fastapi import Depends, Request
from fastapi.responses import StreamingResponse
from genkit.web import to_sse
from genkit_fastapi import handle_genkit_request


# 1. Serve both JSON POST and SSE streaming from a custom route
@app.post('/api/chef')
async def chef_endpoint(request: Request, ctx: dict[str, object] = Depends(user_context)):
    return await handle_genkit_request(request, action=bargain_chef_flow, context=ctx)


# 2. Stream SSE frames from typed FastAPI parameters
@app.post('/api/chef/stream')
async def chef_stream(craving: str, ctx: dict[str, object] = Depends(user_context)):
    stream = bargain_chef_flow.stream(BargainChefInput(craving=craving), context=ctx)
    return StreamingResponse(to_sse(stream), media_type='text/event-stream')

Serve an agent over HTTP

Section titled “Serve an agent over HTTP”

Agents are experimental. To get define_agent, change from genkit import Genkit to from genkit.exp import Genkit in main.py; the rest of the file stays the same. serve_agent mounts the turn route, plus /getSnapshot and /abort when the agent has a session store.

Install genkit-google-cloud for the Firestore-backed store:

Terminal window
uv add genkit-google-cloud
from genkit_fastapi.exp import serve_agent
from genkit_google_cloud.exp import FirestoreSessionStore


bargain_chef_agent = ai.define_agent(
    name='bargainChefAgent',
    model=GoogleAI.gemini('gemini-flash-latest'),
    system='You help users cook from grocery sale ingredients.',
    tools=[get_ingredients_on_sale],
    store=FirestoreSessionStore(),
)


app.include_router(
    serve_agent(
        bargain_chef_agent,
        context_dependency=user_context,
    ),
    prefix='/api',
)
# POST /api/bargainChefAgent
# POST /api/bargainChefAgent/getSnapshot
# POST /api/bargainChefAgent/abort

FirestoreSessionStore uses Application Default Credentials (or FIRESTORE_EMULATOR_HOST locally). See Session stores for options such as tenant prefixes, and Serve agents over HTTP for the turn envelope (data / init), streaming, and client connection details.

Check the project layout

Section titled “Check the project layout”

Verify that your project layout matches the structure below:

  • pyproject.toml
  • main.py

Run the app

Section titled “Run the app”

Start the FastAPI server:

Terminal window
uv run uvicorn main:app --reload

This launches the FastAPI app at http://localhost:8000. With the server running, send a streaming request from another terminal:

Terminal window
curl -N -X POST http://localhost:8000/bargainChefFlow \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer user-123" \
  -H "Accept: text/event-stream" \
  -d '{"data":{"craving":"something warm with chicken"}}'

The { "data": ... } wrapper is required: Genkit’s HTTP handler reads the flow input from the request body’s data field.

The response arrives as a series of data: events. Each event contains the partial recipe accumulated so far: title first, then description, then ingredients (with onSale flags on the ones the model picked from the tool), then steps.

Test and inspect the app

Section titled “Test and inspect the app”

You can test the endpoint directly with curl, and you can use the Developer UI to inspect both manual runs and live requests.

Send a request with curl

Section titled “Send a request with curl”

For a non-streaming response, drop the Accept: text/event-stream header:

Terminal window
curl -X POST http://localhost:8000/bargainChefFlow \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer user-123" \
  -d '{"data":{"craving":"something warm with chicken"}}'

You’ll receive the final structured recipe as a single JSON response.

Use the Developer UI

Section titled “Use the Developer UI”

The Developer UI is Genkit’s local console for testing flows and inspecting execution traces. It runs alongside your backend code, gives you a visual runner for any flow in your project, and records every tool call and model invocation so you can iterate on prompts and debug tool behavior.

  1. Start the Developer UI from your project root:

    Terminal window
    genkit start -- uv run uvicorn main:app --reload

    This launches the Developer UI at http://localhost:4000 by default.

  2. Select bargainChefFlow from the list of flows.

  3. Enter sample input:

    { "craving": "something warm with chicken" }

  4. Click Run.

    You’ll see the generated recipe, with a trace that builds in real time so you can follow the flow’s progress through each tool call and model invocation.

What you built

Section titled “What you built”

You now have a standalone Genkit backend on FastAPI that streams structured output from Gemini over HTTP, calls a tool during generation to ground the model’s response in mock sale-price data, validates input and output against schemas, and surfaces every step in a local trace UI.

Next steps

Section titled “Next steps”