Structured Output + Pydantic in Genkit Python: What Actually Works

You called response.json expecting parsed fields. You set output_format='json' without a schema and got a string back. You have a list model and nothing is parsing. You’re in the right place.

This is the complete guide to structured output in Genkit Python: not only the happy path, but the mistakes that will burn you and the patterns that actually work.

---

The Most Common Mistake (Read This First)

# This is what most developers try first
response = await ai.generate(prompt='Give me user data as JSON.')
data = response.json   # Pydantic's serializer, not parsed schema output

# This is the correct pattern
response = await ai.generate(
    prompt='Give me user data.',
    output_schema=UserData,   # Pydantic model
)
data = response.output   # typed UserData instance

Two things wrong in the first version:

1. response.json is Pydantic’s method for serializing the ModelResponse itself. Prefer response.output for parsed schema data.
2. Without output_schema=, the SDK has nothing to validate against. When you do pass a schema, Genkit defaults output_format to 'json', so you usually don’t need to set the format for single objects.

---

The Correct Pattern: Basic Structured Output

from pydantic import BaseModel
from genkit import Genkit
from genkit_google_genai import GoogleAI

ai = Genkit(
    plugins=[GoogleAI()],              # reads GEMINI_API_KEY from env
    model='googleai/gemini-2.0-flash',
)

class ProductReview(BaseModel):
    product_name: str
    rating: int           # 1-5
    sentiment: str        # positive/negative/neutral
    key_complaint: str

async def extract_review(raw_text: str) -> ProductReview:
    response = await ai.generate(
        prompt=f'Extract review data from this text:\n\n{raw_text}',
        output_schema=ProductReview,
    )
    review = response.output  # ProductReview instance, fully typed
    return review

async def main():
    review = await extract_review(
        "The XPS 15 is amazing but the battery life is terrible. 3/5 stars."
    )
    print(review.product_name)   # "XPS 15"
    print(review.rating)         # 3
    print(review.sentiment)      # "negative" or "neutral"

if __name__ == '__main__':
    ai.run_main(main())   # always ai.run_main(), never asyncio.run()

---

Common Mistakes Table

• Wrong: response.json for parsed fields. Right: response.output. Why: .json serializes the response object, not your schema.
• Wrong: no output_schema=. Right: pass your Pydantic model. Why: without a schema, output is unvalidated text.
• Wrong: assuming you must always pass output_format='json'. Right: omit it for single objects with a schema; set 'array' or 'enum' when you need those formatters. Why: schema alone defaults format to json.
• Wrong: output_format='json' for a list. Right: output_format='array'. Why: different internal formatter.
• Wrong: bare model='gemini-2.0-flash'. Right: model='googleai/gemini-2.0-flash'. Why: plugin prefix required.
• Wrong: asyncio.run(main()). Right: ai.run_main(main()). Why: SDK manages its own event loop.

---

Nested Models

Nested Pydantic models work. Two levels of nesting is the sweet spot. Deeply nested structures (4+ levels) can cause the model to miss fields.

from pydantic import BaseModel
from typing import Optional

class Address(BaseModel):
    street: str
    city: str
    country: str

class Company(BaseModel):
    name: str
    founded_year: int
    headquarters: Address          # nested model
    employee_count: Optional[int]  # model may not fill this

async def extract_company(text: str) -> Company:
    response = await ai.generate(
        prompt=f'Extract company information:\n\n{text}',
        output_schema=Company,
    )
    company = response.output
    # company.headquarters is an Address instance
    print(company.headquarters.city)
    return company

Genkit serializes the full Pydantic JSON schema and injects it into the prompt as a system instruction. The model sees the complete nested structure. For deeply nested schemas, add a system prompt hint: “Fill all fields including nested objects.”

---

Lists of Structured Objects

For lists, the output_format changes to 'array' and you pass a JSON schema dict via TypeAdapter:

from pydantic import BaseModel, TypeAdapter

class Ingredient(BaseModel):
    name: str
    quantity: str
    unit: str

async def extract_ingredients(recipe_text: str) -> list[Ingredient]:
    # Build the array schema
    schema = TypeAdapter(list[Ingredient]).json_schema()

    response = await ai.generate(
        prompt=f'Extract all ingredients from this recipe:\n\n{recipe_text}',
        output_format='array',
        output_schema=schema,          # dict, not the class
    )
    # response.output is a list of dicts here, not Ingredient instances
    ingredients_raw = response.output
    # Validate manually if you need typed objects:
    return [Ingredient(**item) for item in ingredients_raw]

When using output_format='array' with a schema dict, response.output returns a list of dicts, not Pydantic instances. Call Ingredient(**item) to get typed objects.

---

Enums for Classification

output_format='enum' is the cleanest way to do classification:

from enum import Enum
from pydantic import BaseModel

class Sentiment(str, Enum):
    POSITIVE = 'positive'
    NEGATIVE = 'negative'
    NEUTRAL = 'neutral'

async def classify_sentiment(text: str) -> str:
    response = await ai.generate(
        prompt=f'Classify the sentiment of this text: {text}',
        output_format='enum',
        output_schema={
            'type': 'string',
            'enum': [e.value for e in Sentiment]
        },
    )
    # response.output is a raw string: 'positive', 'negative', or 'neutral'
    result = response.output
    return Sentiment(result)   # convert to enum

async def main():
    label = await classify_sentiment("This product exceeded my expectations!")
    print(label)          # Sentiment.POSITIVE
    print(label.value)    # 'positive'

The enum formatter strips surrounding quotes and validates the model returned one of the allowed values. If it doesn’t match, you’ll get a GenkitError.

---

Optional Fields and Defaults

This is where most production code breaks. Models frequently skip optional fields. Always provide defaults:

from pydantic import BaseModel, Field
from typing import Optional

class ArticleMetadata(BaseModel):
    title: str
    author: Optional[str] = None         # model may leave this blank
    word_count: Optional[int] = None
    tags: list[str] = Field(default_factory=list)
    is_paywalled: bool = False
    confidence: float = Field(default=0.0, ge=0.0, le=1.0)

async def extract_metadata(article: str) -> ArticleMetadata:
    response = await ai.generate(
        prompt=f'Extract article metadata. Use null for unknown fields.\n\n{article}',
        output_schema=ArticleMetadata,
    )
    return response.output  # Optional fields default safely

async def main():
    meta = await extract_metadata("Breaking: AI startup raises $50M...")
    print(meta.title)          # filled
    print(meta.author)         # None if model skipped it; no KeyError
    print(meta.tags)           # [] if model skipped it; no AttributeError

Use Optional[T] = None for fields the model might skip. Use Field(default_factory=list) for list fields. Never assume the model will fill every field; it won’t.

---

Validation Errors and Graceful Handling

When the model returns malformed JSON, Genkit raises a GenkitError. You should always wrap structured output calls with a fallback:

from genkit import GenkitError

async def safe_extract(text: str) -> ProductReview | None:
    try:
        response = await ai.generate(
            prompt=f'Extract review data:\n\n{text}',
            output_schema=ProductReview,
        )
        return response.output
    except GenkitError as e:
        # GenkitError carries a status code (INVALID_ARGUMENT, INTERNAL, etc.)
        print(f'Extraction failed: {e.status}: {e.message}')
        return None
    except Exception as e:
        # Network errors, model unavailable, etc.
        print(f'Unexpected error: {e}')
        return None

async def main():
    result = await safe_extract("Ambiguous text that might not parse...")
    if result is None:
        # fallback: retry with a more explicit prompt or use unstructured output
        print("Extraction failed; using fallback")
    else:
        print(result.product_name)

When validation fails, Genkit’s error will include the raw model output in the message field. That’s useful for debugging. Log e.message to see what the model actually returned.

---

Streaming Structured Output

Short answer: streaming and structured JSON output do not mix cleanly.

generate_stream accepts output_format and output_schema, but chunks arrive as partial text. The output field on each chunk is not a valid parsed object until the stream is complete.

# The right way to get structured output from a stream
async def stream_with_structured_final():
    sr = ai.generate_stream(
        prompt='List 5 programming languages with year created.',
        output_format='array',
        output_schema=TypeAdapter(list[Language]).json_schema(),
    )

    # Show chunks to the user as they arrive (raw text)
    async for chunk in sr.stream:
        if chunk.text:
            print(chunk.text, end='', flush=True)

    # Get the final response; this is the parsed structured output
    final = await sr.response
    languages = final.output   # list of dicts, parsed from the completed stream
    return languages

The SDK parses the accumulated stream at the end. So you get streaming display with a structured final result, which is usually exactly what you want.

Do not try to parse chunk.output mid-stream for a JSON schema. The intermediate output is incomplete JSON.

---

Complete Example: Classification + Extraction Pipeline

Here’s everything in one pattern: a content moderation pipeline that classifies and then extracts structured detail.

from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field
from genkit import Genkit, GenkitError
from genkit_google_genai import GoogleAI

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

# Step 1: Classify
class ContentCategory(str, Enum):
    SAFE = 'safe'
    SPAM = 'spam'
    HARMFUL = 'harmful'

# Step 2: Extract detail (only for non-safe content)
class ModerationDetail(BaseModel):
    category: str
    reason: str
    severity: int = Field(ge=1, le=5)
    flagged_phrases: list[str] = Field(default_factory=list)
    recommended_action: Optional[str] = None

class ModerationResult(BaseModel):
    category: ContentCategory
    detail: Optional[ModerationDetail] = None

@ai.flow()
async def moderate_content(content: str) -> ModerationResult:
    # Phase 1: classify
    try:
        classify_response = await ai.generate(
            prompt=f'Classify this content. Return ONLY one word: safe, spam, or harmful.\n\n{content}',
            output_format='enum',
            output_schema={
                'type': 'string',
                'enum': ['safe', 'spam', 'harmful']
            },
        )
        category = ContentCategory(classify_response.output)
    except (GenkitError, ValueError):
        category = ContentCategory.SAFE  # safe default on failure

    # Phase 2: extract detail only for flagged content
    if category == ContentCategory.SAFE:
        return ModerationResult(category=category, detail=None)

    try:
        detail_response = await ai.generate(
            prompt=f'This content was classified as {category.value}. Extract moderation details.\n\n{content}',
            output_schema=ModerationDetail,
        )
        detail = detail_response.output
    except GenkitError as e:
        # If extraction fails, return category without detail
        print(f'Detail extraction failed: {e.message}')
        detail = None

    return ModerationResult(category=category, detail=detail)

async def main():
    test_cases = [
        "Great product! Highly recommend.",
        "BUY NOW! CLICK HERE! FREE MONEY!!!",
        "I want to harm someone at the local school.",
    ]

    for content in test_cases:
        result = await moderate_content(content)
        print(f'\n--- Content: {content[:40]}...')
        print(f'Category: {result.category.value}')
        if result.detail:
            print(f'Severity: {result.detail.severity}/5')
            print(f'Reason: {result.detail.reason}')
            print(f'Action: {result.detail.recommended_action}')

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

This pipeline demonstrates the full set:

• output_format='enum' for classification
• output_schema with a typed Pydantic model (format defaults to json)
• Optional fields with defaults
• GenkitError handling with fallbacks
• @ai.flow() for automatic tracing
• ai.run_main() for correct event loop management

---

Common Mistakes Table (Extended)

• Mistake: response.json. What happens: you get a serializer (or a dump of the whole response), not schema fields. Fix: use response.output.
• Mistake: output_format='json' without output_schema=. What happens: unvalidated string. Fix: pass the schema; format is optional for single objects.
• Mistake: output_format='json' for a list. What happens: parser fails. Fix: use output_format='array'.
• Mistake: TypeAdapter schema with output_format='json'. What happens: schema mismatch. Fix: use output_format='array'.
• Mistake: required fields on nested models. What happens: validation error when the model skips. Fix: make fields Optional[T] = None.
• Mistake: asyncio.run(main()). What happens: event loop conflict. Fix: ai.run_main(main()).
• Mistake: parsing chunk.output mid-stream. What happens: incomplete JSON. Fix: only use final.output after await sr.response.
• Mistake: no GenkitError handling. What happens: unhandled exception in prod. Fix: wrap structured output calls.

---

Quick Reference

# Single object (output_format defaults to 'json' when schema is set)
response = await ai.generate(
    prompt='...',
    output_schema=MyModel,    # Pydantic class
)
obj = response.output         # MyModel instance

# List of objects
from pydantic import TypeAdapter
schema = TypeAdapter(list[MyModel]).json_schema()
response = await ai.generate(
    prompt='...',
    output_format='array',
    output_schema=schema,     # dict
)
items = [MyModel(**i) for i in response.output]  # list of dicts → typed

# Enum classification
response = await ai.generate(
    prompt='...',
    output_format='enum',
    output_schema={'type': 'string', 'enum': ['a', 'b', 'c']},
)
label = response.output  # 'a', 'b', or 'c' (raw string)

# Streaming (display chunks, get structured final)
sr = ai.generate_stream(prompt='...', output_schema=MyModel)
async for chunk in sr.stream:
    print(chunk.text, end='')
final = await sr.response
obj = final.output  # MyModel, parsed after stream completes

---

Install

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

Set GEMINI_API_KEY from Google AI Studio.

---

The patterns in this article are verified against Genkit Python 0.8.1 on PyPI (genkit + genkit-google-genai).