developing-genkit-python

Genkit Python

Build AI features in Python — generation, structured output, streaming, tools, flows, and multi-turn agents — with Genkit Python 0.13.

• New project? See Setup and Examples.
• Upgrading from 0.12 or fixing broken 0.12 imports/signatures? Read Migrating from 0.12 to 0.13 first.

Prerequisites

• Python 3.10+ and uv (install)
• Genkit CLI: npm install -g genkit-cli if genkit --version is missing

Hello World

from genkit import Genkit
from genkit_google_genai import GoogleAI

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


async def main() -> None:
    response = await ai.generate(prompt='Suggest one seasonal pasta dish.')
    if response.error:
        raise RuntimeError(response.error.message)
    print(response.text)
    # => Lemony tagliatelle with asparagus, sweet peas, and herb ricotta


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

ai.generate does not raise when the provider rejects or fails a call (INVALID_ARGUMENT, UNAVAILABLE, rate limits). It returns response.error (Failure with .status and .message), finish_reason == 'failed', and empty text. Problems caught before the call raise: GoogleAI() raises ValueError without GEMINI_API_KEY; an unknown model, invalid key, or unknown config field raises GenkitError. See Common Errors.

Agents (Experimental)

Multi-turn chats with history, typed state, human approval, branching, and background work live under genkit.exp and genkit.exp.agent. Start here: Agents.

from genkit.exp import Genkit
from genkit.exp.agent import InMemorySessionStore
from genkit_google_genai import GoogleAI

ai = Genkit(plugins=[GoogleAI()], model=GoogleAI.gemini('gemini-flash-latest'))
agent = ai.define_agent(
    name='concierge',
    system='You are a concise hotel concierge.',
    store=InMemorySessionStore(),
)

chat = agent.chat()
res = await chat.send('Recommend a quiet dinner spot.')  # AgentResponse
print(res.text)
# => Try Osteria Lume on 5th; it has booths and low music.

turn = chat.send_stream('Is it open past 10pm?')  # AgentTurn
async for chunk in turn.stream:
    if chunk.text:
        print(chunk.text, end='')
final = await turn.response  # AgentResponse

More: sessions · HITL · branching · background · state · artifacts · custom · HTTP

Imports

Every symbol lives in one public module (never import private _ submodules; see Migrating from 0.12 to 0.13 if a 0.12 symbol moved):

• Core app: from genkit import Genkit, FlowContext, ToolContext, Message, Part, Role, Media, Document, MultipartToolResponse, PublicError, GenkitError, Failure, ErrorReason
• Google GenAI & Vertex AI: from genkit_google_genai import GoogleAI, VertexAI, GeminiConfig, GeminiImageConfig, GeminiTtsConfig
• Other providers: from genkit_vertexai import ModelGarden · from genkit_anthropic import Anthropic, AnthropicConfig · from genkit_openai import OpenAI, OpenAIConfig · from genkit_ollama import Ollama, OllamaConfig · from genkit_amazon_bedrock import Bedrock, BedrockConfig
• Built-in middleware: from genkit_middleware import Middleware, Retry, Fallback, ToolApproval, Skills, Filesystem (plus from genkit_middleware.exp import Artifacts for agent sessions)
• Writing custom middleware: from genkit.middleware import Middleware, MiddlewareContext
• FastAPI & HTTP: from genkit_fastapi import serve_flow, genkit_fastapi_handler, handle_genkit_request · from genkit_fastapi.exp import serve_agent · from genkit.web import RequestData, to_sse, error_body, error_status
• Agents (experimental): from genkit.exp import Genkit, current_session · from genkit.exp.agent import Agent, Session, InMemorySessionStore, FileSessionStore, remote_agent
• Telemetry: from genkit_google_cloud import enable_google_cloud_telemetry · from genkit.telemetry import configure_instrumentation · from genkit_otel import GenAiInstrumentation
• MCP (Client & Host): from genkit_mcp import define_mcp_client, define_mcp_host, McpClient, McpHost
• Testing & Mocks: from genkit.testing import define_echo_model, define_scripted_model, EchoModel, ScriptedModel
• Evaluators (0.13 only; cut from 1.0, additive later): from genkit_evaluators import register_genkit_evaluators · from genkit.evaluator import BaseDataPoint, EvaluatorResponse, Score, ScoreDetails, ScoreStatus

Workflow

1. Agent or flow? Use ai.define_agent (on genkit.exp.Genkit, see Agents) for multi-turn chats, state, or human-in-the-loop approval instead of hand-rolling a message loop inside a flow. Use @ai.flow() on genkit.Genkit for stateless request/response or streaming pipelines.
2. Configure credentials & model refs. Set GEMINI_API_KEY. Prefer typed refs (GoogleAI.gemini('gemini-flash-latest')) and typed flat configs (GeminiConfig(...)) so your type checker validates fields before runtime. Prefixed strings ('googleai/gemini-flash-latest') also work.
3. Enter via ai.run_main(main()) for standalone scripts and CLI apps so the process stays up for the Dev UI under genkit start. FastAPI apps skip run_main; uvicorn keeps the process alive: genkit start -- uv run uvicorn main:app --reload.
4. Run with Dev Workflow (genkit start + Dev UI) and type-check with pyright, mypy, or ty.
5. Verify with traces, not a blind uv run. See Genkit CLI.
6. Stuck? Check Gotchas, Common Errors, or Migrating from 0.12 to 0.13.

Gotchas

• Agents need genkit.exp: from genkit.exp import Genkit. Stable genkit.Genkit has no define_agent.
• Tools are async: @ai.tool() raises TypeError on a plain def; use async def.
• Check response.error: provider failures come back on the response, not as exceptions (see Hello World).
• Gemini thinking levels: GeminiConfig(thinking_level='MINIMAL') fails on gemini-flash-latest; use 'LOW', 'MEDIUM', or 'HIGH'.
• Vertex AI model IDs: Vertex does not serve -latest aliases. Pass a pinned model ID to VertexAI.gemini(...).
• Logging: genkit.get_logger is structlog-based. enable_google_cloud_telemetry() injects trace context for GCP log-trace correlation.
• Cloud telemetry auth: call enable_google_cloud_telemetry() with Application Default Credentials or explicit credentials= (service account dict or Credentials instance).

Genkit CLI (recommended)

genkit start wraps any Python program that uses Genkit, capturing traces from every action while forwarding stdio cleanly. Running uv run directly skips trace capture.

Primary pattern (persistent server + Dev UI):

genkit start -- uv run src/main.py
genkit start --noui -- uv run src/main.py   # persistent runtime without opening the Dev UI

genkit start runs until stopped with Ctrl+C (even with --noui). Do not run genkit start as a blocking step in automated/non-interactive scripts; use flow:run instead.

Non-interactive use (agents/CI): pass --non-interactive before -- so the CLI never blocks on prompts:

genkit start --non-interactive -- uv run src/main.py

One-shot flow execution (flow:run): spins up the runtime, invokes one @ai.flow() by name, prints a Trace ID, and exits:

genkit flow:run myFlow '{"data": "input"}' -- uv run src/main.py

Note: flow:run invokes flows (@ai.flow()), not agents (ai.define_agent). Wrap a single agent turn in a helper flow to exercise it via flow:run (see Agents).

Inspecting traces:

genkit trace:list                        # list recent trace IDs
genkit trace:get <traceId>               # human-readable trace inspection
genkit trace:get <traceId> --format json # machine-readable JSON for jq / scripts

Always pass --format json when piping into jq or other parsers.

See Dev Workflow for the full pre-run checklist and Dev UI walkthrough.

References

• Migrating from 0.12 to 0.13: Deterministic 0.12 -> 0.13 renames, config flattening, PublicError, and exp moves.
• Examples: Structured output, streaming, media parts, flows, tools, interrupts, cancellation, embeddings.
• Setup: Project bootstrap with uv, version alignment, and provider plugins.
• Common Errors: Fast fixes for 0.13 validation errors, HTTP 500 redaction, streaming, and sessions.
• FastAPI: serve_flow, serve_agent, PublicError, SSE streaming, and parallel flows.
• Dotprompt: .prompt files, schema validation, variants, partials, and helpers.
• Evals: 0.13 evaluator syntax (deprecated; removed in 1.0).
• Dev Workflow: genkit start, Dev UI, and trace verification checklist.
• Agents (Experimental): Multi-turn stateful agents (genkit.exp).