Genkit Python in 2026: The Definitive Getting-Started Guide

Here’s a working Genkit Python app in under 20 lines:

from genkit import Genkit
from genkit_google_genai import GoogleAI

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

async def main():
    response = await ai.generate(prompt='Explain async/await in one sentence.')
    print(response.text)

if __name__ == '__main__':
    ai.run_main(main())

That’s the skeleton. Every Genkit Python app follows this shape. The rest of this guide fills in the details: installation, flows, structured output, async patterns, and a complete end-to-end example. If you already tried the JS version and want to know what’s different in Python, I’ll call that out explicitly.

Why Genkit Python instead of the Vertex AI SDK directly?

You could use google-generativeai or vertexai directly. Here’s why you’d reach for Genkit instead:

Observability out of the box. Every generate() call and flow execution is traced. You get a local Dev UI that shows you request/response pairs, token counts, latencies, and span trees without writing any logging code. In production, this hooks into OpenTelemetry.

Flows are deployable units. A flow is a typed, observable async function that the framework knows how to call and test. You can expose it over HTTP from your own FastAPI or Flask app; Genkit itself stays focused on the AI runtime.

Middleware for request hooks. Genkit ships BaseMiddleware and a use=[...] slot on generate. You write your own retry, logging, or approval logic as middleware rather than wrapping every call by hand. Named helpers like Retry or Fallback are not in the 0.8.1 package; the extension point is.

JS parity. If your team has a JS Genkit app and you’re adding a Python backend (or vice versa), the mental model is identical: flows, tools, structured output, same Dev UI.

The tradeoff: Genkit Python adds indirection. If you just need a one-shot LLM call with no tracing or observability, the raw SDK is simpler. Genkit earns its keep at scale.

Installation

The quick way with `uv` (recommended)

# Create a project
mkdir my-genkit-app && cd my-genkit-app
uv init --python 3.12
uv add "genkit[google-genai]==0.8.1"

That pulls genkit and genkit-google-genai from PyPI together via the [google-genai] extra.

With pip (via uv)

uv pip install "genkit[google-genai]==0.8.1"

API key

export GEMINI_API_KEY="your-key-here"

GoogleAI() reads GEMINI_API_KEY by default. You can also pass GoogleAI(api_key='...') if you prefer not to use the env var.

Init: one object rules everything

from genkit import Genkit
from genkit_google_genai import GoogleAI

ai = Genkit(
    plugins=[GoogleAI()],          # load model providers
    model='googleai/gemini-2.0-flash',  # default model for all calls
)

Model ID format matters. The model string is always "googleai/<model-name>". Never a bare model name. Common options:

googleai/gemini-2.0-flash (Gemini 2.0 Flash): default, fast and cheap
googleai/gemini-flash-latest (Gemini 2.5 Flash): balanced quality/speed
googleai/gemini-2.5-pro (Gemini 2.5 Pro): best quality, slower

The ai object is your app’s runtime. It holds the plugin registry, middleware registry, and async event loop. You create one and use it everywhere.

First `generate()` call

response = await ai.generate(prompt='Write a haiku about Python.')
print(response.text)

Output:

Indented stanzas,
whitespace speaks louder than words.
Guido smiles softly.

Two things to know about response:

response.text: the model’s text output (string)
response.output: structured output when you’ve specified a schema (more on this below)
response.messages: the full conversation history including this turn

Never use asyncio.run() to call async Genkit code. Use ai.run_main():

async def main():
    response = await ai.generate(prompt='Hello')
    print(response.text)

if __name__ == '__main__':
    ai.run_main(main())  # correct
    # asyncio.run(main())  # breaks the reflection server

ai.run_main() starts your coroutine and, in dev mode, keeps the process alive to serve the Dev UI’s reflection server.

Defining a Flow

A flow is the core Genkit primitive. It’s an async function with:

Typed input/output (Pydantic models recommended)
Automatic tracing
Callable from the Dev UI
Easy to wrap behind your own HTTP routes (FastAPI, Flask, or anything else)

from pydantic import BaseModel

class SummarizeInput(BaseModel):
    text: str
    max_words: int = 50

@ai.flow()
async def summarize(input: SummarizeInput) -> str:
    response = await ai.generate(
        prompt=f'Summarize in at most {input.max_words} words: {input.text}'
    )
    return response.text

Call it like a normal async function:

result = await summarize(SummarizeInput(
    text='Large language models are neural networks trained...',
    max_words=20,
))
print(result)
# "Large language models are neural networks trained on text to generate language."

The @ai.flow() decorator registers the function with the Genkit runtime. You can give it a custom name:

@ai.flow(name='my-summarizer')
async def summarize(input: SummarizeInput) -> str:
    ...

Structured Output with Pydantic

This is where Genkit Python shines. You define a Pydantic model, pass it as output_schema, and get a typed object back with no manual JSON parsing.

from pydantic import BaseModel

class BookReview(BaseModel):
    title: str
    author: str
    rating: int       # 1-5
    summary: str
    recommend: bool

response = await ai.generate(
    prompt='Review "The Pragmatic Programmer" by David Thomas and Andrew Hunt.',
    output_schema=BookReview,
)

review = response.output   # BookReview instance
print(f"{review.title}: {review.rating}/5 stars")
print(f"Recommend: {review.recommend}")
print(review.summary)

When output_schema is set, Genkit defaults output_format to 'json', so you can omit it for single objects. Pass output_format explicitly when you need 'array', 'enum', or another formatter.

Use response.output, not response.json. On a ModelResponse, .json is Pydantic’s serializer for the response object itself, not the parsed schema.

For lists of structured objects, use Pydantic’s TypeAdapter and output_format='array':

from pydantic import TypeAdapter

schema = TypeAdapter(list[BookReview]).json_schema()
response = await ai.generate(
    prompt='Give me reviews of 3 classic programming books.',
    output_format='array',
    output_schema=schema,
)
books = response.output  # list of dicts

Available output_format values: 'text', 'json', 'array', 'enum', 'jsonl'.

Streaming

Streaming lets you display partial output as it arrives, which matters for any user-facing app.

# generate_stream is NOT awaited; it returns synchronously
sr = ai.generate_stream(prompt='Write a short story about a robot learning to code.')

async for chunk in sr.stream:
    if chunk.text:
        print(chunk.text, end='', flush=True)

final = await sr.response   # full ModelResponse when done
print(f"\n\nTotal tokens: {final.usage.total_tokens}")

Note the asymmetry: generate() is awaited; generate_stream() is not. It returns immediately with a stream object. You then iterate .stream async, and await .response to get the final result.

Streaming flows

Flows can stream chunks to callers using ActionRunContext:

from genkit import ActionRunContext

@ai.flow()
async def stream_story(subject: str, ctx: ActionRunContext) -> str:
    sr = ai.generate_stream(prompt=f'Write a 3-paragraph story about {subject}.')
    full_text = ''
    async for chunk in sr.stream:
        if chunk.text:
            ctx.send_chunk(chunk.text)
            full_text += chunk.text
    return full_text

The ctx parameter is injected by the framework when present. You don’t pass it when calling the flow.

Running Locally with the Dev Server

The Dev UI is where development actually happens. It lets you call flows interactively, inspect traces, and test structured output without writing a test harness.

# Start the dev server (from your project directory)
genkit start -- uv run src/main.py

This starts your Python app with the Genkit reflection server enabled, then opens the Dev UI at http://localhost:4000. You’ll see all your registered flows in a sidebar. Click any flow, fill in the input JSON, and run it. Every call shows up as a trace with full request/response details.

If you’re using pip instead of uv:

genkit start -- python src/main.py

Install the Genkit CLI with: npm install -g genkit-cli

Complete End-to-End Example: Article Summarizer

Here’s a complete, working app that combines flows, structured output, and tools:

# src/main.py
from pydantic import BaseModel, Field
from genkit import Genkit
from genkit_google_genai import GoogleAI

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


# --- Schemas ---

class ArticleInput(BaseModel):
    url: str = Field(description='Article URL to summarize')
    audience: str = Field(
        default='general',
        description='Target audience: general, technical, executive',
    )

class ArticleSummary(BaseModel):
    title: str
    one_liner: str
    key_points: list[str]
    sentiment: str   # positive, negative, neutral
    reading_time_minutes: int


# --- Tool: fetch article text ---

class FetchInput(BaseModel):
    url: str

@ai.tool()
async def fetch_article(input: FetchInput) -> str:
    """Fetch the text content of a URL."""
    import urllib.request
    try:
        with urllib.request.urlopen(input.url, timeout=10) as resp:
            # In production: use httpx + HTML stripping
            return resp.read().decode('utf-8')[:8000]
    except Exception as e:
        return f'Could not fetch article: {e}'


# --- Flow ---

@ai.flow()
async def summarize_article(input: ArticleInput) -> ArticleSummary:
    """Fetch an article and return a structured summary."""

    # Step 1: Fetch the article using tool use
    fetch_response = await ai.generate(
        prompt=f'Fetch the article at {input.url}',
        tools=[fetch_article],
    )

    article_text = fetch_response.text

    # Step 2: Summarize with structured output
    summary_response = await ai.generate(
        prompt=f"""
Summarize the following article for a {input.audience} audience.
Article:
{article_text}
""",
        output_schema=ArticleSummary,
    )

    return summary_response.output


async def main():
    result = await summarize_article(ArticleInput(
        url='https://example.com/some-article',
        audience='technical',
    ))
    print(f"Title: {result.title}")
    print(f"One-liner: {result.one_liner}")
    print(f"Sentiment: {result.sentiment}")
    print("Key points:")
    for point in result.key_points:
        print(f"  • {point}")


if __name__ == '__main__':
    ai.run_main(main())

Run it:

uv run src/main.py

Or explore it interactively:

genkit start -- uv run src/main.py

Then open http://localhost:4000, click summarize_article, and paste in any URL.

Common Mistakes Quick Reference

Wrong: model='gemini-2.0-flash'. Right: model='googleai/gemini-2.0-flash'. Why: must include the provider prefix.
Wrong: await ai.generate_stream(...). Right: ai.generate_stream(...) (no await). Why: returns a sync stream object.
Wrong: asyncio.run(main()). Right: ai.run_main(main()). Why: keeps the reflection server working in dev.
Wrong: response.json for parsed fields. Right: response.output. Why: .json is Pydantic serialization of the response, not schema output.
Wrong: response.message for text. Right: response.text. Why: correct attribute name.
Wrong: assuming you must pass api_key= to GoogleAI. Right: set GEMINI_API_KEY, or pass api_key= if you prefer. Why: both work.
Wrong: @ai.define_tool(). Right: @ai.tool(). Why: correct decorator name.
Wrong: async def tool(city: str). Right: input as a Pydantic BaseModel. Why: Gemini expects an OBJECT-shaped tool schema.

What’s Next

Things that already ship in Genkit Python 0.8.1 and are worth learning next:

Interrupt helpers (define_interrupt, respond_to_interrupt, restart_tool) for pausing a tool call and resuming later
ai.embed / ai.embed_many when you need RAG or similarity search
Passing retrieved context with docs= on generate
Custom middleware via BaseMiddleware and use=[...] when you need request/response hooks

The full SDK source is at github.com/genkit-ai/genkit in py/. File issues there; join the discussion on Discord.