Structured Output + Pydantic in Genkit Python: What Actually Works

You called response.json and got None. You set output_format='json' 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 the happy path, but the full picture including 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   # AttributeError or None

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

Two things wrong in the ❌ version:

1. response.json doesn’t exist the way you think it does — use response.output
2. Without output_format='json' + output_schema=, you’re asking the model to return JSON but not instructing the SDK to parse it

---

The Correct Pattern: Basic Structured Output

from pydantic import BaseModel
from genkit import Genkit
from genkit.plugins.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_format='json',
        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 · ✅ Right: response.output · Why: .json doesn’t hold parsed output
• ❌ Wrong: response.message · ✅ Right: response.text · Why: Wrong attribute for text
• ❌ Wrong: No output_format= · ✅ Right: Always pass output_format=‘json’ · Why: SDK won’t parse without it
• ❌ 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
• ❌ Wrong: Forgetting output_schema= · ✅ Right: Pass your Pydantic model · Why: Without schema, output is unvalidated

---

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_format='json',
        output_schema=Company,
    )
    company = response.output
    # company.headquarters is an Address instance
    print(company.headquarters.city)
    return company

Key insight: 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]

Important: 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_format='json',
        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

Pattern: 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._core._error import GenkitError

async def safe_extract(text: str) -> ProductReview | None:
    try:
        response = await ai.generate(
            prompt=f'Extract review data:\n\n{text}',
            output_format='json',
            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 uses extract_json_array_from_text internally to parse the accumulated stream at the end. So you get streaming display with 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:

import asyncio
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field
from genkit import Genkit
from genkit._core._error import GenkitError
from genkit.plugins.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_format='json',
            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_format='json' with a typed Pydantic schema
• ✅ 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: None or AttributeError · Fix: Use response.output
• Mistake: output_format=‘json’ without output_schema= · What Happens: Unvalidated string · Fix: Always pair them
• 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: ValidationError when 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: Always wrap structured output calls

---

Quick Reference

# Single object
response = await ai.generate(
    prompt='...',
    output_format='json',
    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_format='json', 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] @ git+https://github.com/firebase/genkit.git#subdirectory=py/packages/genkit"

Plugins are not on PyPI yet — install from git. Set GEMINI_API_KEY from Google AI Studio.

---

The patterns in this article are verified against the Genkit Python main branch (June 2026). API surface verified against source at py/packages/genkit/src/genkit/_ai/_formats/.