Genkit Python SDK v0.12.0 Release Notes

Genkit Python SDK v0.12.0 is here. This release is about A2UI surfaces, Deep Research (plus Antigravity and Lyria), a stronger OpenAI plugin, and generate() handing back a ModelResponse once a turn has started.

uv add genkit genkit-google-genai

What's New

A2UI surfaces (#6312)

uv add genkit-a2ui

Surfaces() watches generate for ```a2ui fences and rewrites them into application/a2ui+json data parts. On the next turn those parts become text again, so the model can see the surface it already drew and the button the user clicked.

from genkit import Genkit
from genkit_a2ui import Surfaces, envelopes_from_parts
from genkit_google_genai import GoogleAI

ai = Genkit(plugins=[GoogleAI()])

response = await ai.generate(
    model=GoogleAI.gemini_model('gemini-flash-latest'),
    prompt='Show me the weather in Tokyo.',
    use=[Surfaces()],
)
print(envelopes_from_parts(response.message.content))

Surfaces() uses the bundled catalog. load_catalog(ai, catalog) plus Surfaces(catalog=catalog.id) uses yours. An unregistered catalog id, or validate='strict' on a surface that does not match, fails the turn: finish_reason is failed and .message is None.

Deep Research, Antigravity, and Lyria (#6240, #6241)

Deep Research is a job, like Veo. generate_operation starts it. check_operation is how you find out when the report is ready. A finished report can take several minutes.

from genkit import Genkit
from genkit_google_genai import GoogleAI

ai = Genkit(plugins=[GoogleAI()])

operation = await ai.generate_operation(
    model=GoogleAI.deep_research_model('deep-research-preview-04-2026'),
    prompt='Summarize recent advances in quantum error correction for a technical audience.',
)
while not operation.done:
    operation = await ai.check_operation(operation)
print(operation.output)

Also on GoogleAI: deep-research-max-preview-04-2026 and deep-research-pro-preview-12-2025.

Antigravity and Lyria are ordinary generate calls on the same plugin. These three are Google AI only. Vertex does not resolve these ids.

reply = await ai.generate(
    model=GoogleAI.antigravity_model('antigravity-preview-05-2026'),
    prompt='Draft a weekend in Paris.',
)
print(reply.text)

clip = await ai.generate(
    model=GoogleAI.lyria_model('lyria-3-clip-preview'),
    prompt='A short piano loop, rainy window.',
)
for part in clip.message.content:
    if part.media:
        print(part.media.url)

lyria-3-pro-preview is the other Lyria id.

OpenAI plugin, including gpt-6-astra (#6306)

gpt-6-astra is in the catalog. It advertises media, a system role, and JSON. It does not advertise tools.

from genkit import Genkit
from genkit_openai import OpenAI

ai = Genkit(plugins=[OpenAI()])

response = await ai.generate(
    model=OpenAI.gpt_model('gpt-6-astra'),
    prompt='Say hi in one word.',
)
print(response.text)

Streaming now emits content, reasoning, and tool-call deltas that arrive in the same chunk (#6282). Responses carry finish reason, finish message, and refusals (#6259). raw and custom metadata are set on the response (#6261). A completion with no choices no longer crashes (#6258). Embeddings requested as base64 validate (#6346). Reasoning models get max_tokens mapped (#6324). A non-streaming call no longer sends stream options (#6297). Embedder and list_actions HTTP errors arrive as GenkitError (#6225).

generate() raises only when a turn never starts (#6255)

If generate() stops before a turn starts, it raises. Once the model has been called, you get a ModelResponse, including when that turn dies. This is extremely helpful when you want to retry from the rounds that already finished.

from pydantic import BaseModel
from genkit import Genkit, GenkitError, RuntimeErrorReason
from genkit_google_genai import GoogleAI

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


class Recipe(BaseModel):
    title: str


try:
    await ai.generate(model='nope/x', prompt='hi')
except GenkitError as err:
    print(err.reason)  # MODEL_NOT_FOUND

response = await ai.generate(
    prompt='Give me a recipe as JSON.',
    output_schema=Recipe,
)
if response.output is not None:
    print(response.output.title)
else:
    print(response.error.reason, response.text)

A turn never starts, and generate() raises, for an unknown model, tool, or format; two tools with the same name; max_turns < 0; or a config class that does not belong to the model.

Once the model has been called, the reason sits on response.error and .messages is what you can send again. Text that is not a Recipe stays finish_reason stop, .output is None, .text is the raw reply, and error.reason is INVALID_OUTPUT. In v0.11.0 that case was finish_reason failed.

A turn that dies is failed, and .message is None: the model named a tool that is not registered (TOOL_NOT_FOUND), or a plugin refused the output. Hitting max_turns is aborted, and the unanswered tool request is dropped.

raise PublicError('NOT_FOUND', 'No order 123') inside a tool publishes that sentence on response.error.message. A bare exception reads internal error. abort_signal returns an aborted response, including when you set it before the first model call. asyncio.wait_for and task.cancel() still raise.

Typed stream chunks (#6212)

generate_stream(..., output_schema=Country) used to give you a dict you indexed by hand. After this, each chunk.output is a Country. This is extremely helpful when you want to paint a title as soon as that field exists.

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

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


class Country(BaseModel):
    name: str
    capital: str
    population: int


sr = ai.generate_stream(
    prompt='Give quick facts about Japan.',
    output_schema=Country,
)
async for chunk in sr:
    if chunk.output and chunk.output.name:
        print(chunk.output.name)

print((await sr.response).output)

Before the JSON object starts, chunk.output is None. Once it has started, you get a Country. Fields that have not arrived are None. A string cut mid-token is the prefix ("Ja"). Constraints and validators are skipped on chunks. (await sr.response).output is the finished, validated object. A dict schema, or no schema, still gives you a dict. define_prompt(..., output_schema=Country).stream() works the same way.

Existing API Changes

Agents are on genkit.exp (#6313)

Agents moved to the experimental namespace. They will graduate to stable over time.

# OLD (v0.11.0):
from genkit import Genkit
from genkit.agent import InMemorySessionStore

# NEW (v0.12.0):
from genkit.exp import Genkit, InMemorySessionStore

One Part (#6243)

The part API is one Part. Walking the messages on a model response, you tell the kinds apart by which field is set.

# OLD (v0.11.0):
from genkit import MediaPart, TextPart, ToolRequestPart, ToolResponsePart

for message in response.messages:
    for part in message.content:
        root = part.root
        if isinstance(root, TextPart):
            print(root.text)
        elif isinstance(root, MediaPart):
            print(root.media.url)
        elif isinstance(root, ToolRequestPart):
            print(root.tool_request.name)
        elif isinstance(root, ToolResponsePart):
            print(root.tool_response.output)

# NEW (v0.12.0):
from genkit import Part

for message in response.messages:
    for part in message.content:
        if part.text is not None:
            print(part.text)
        elif part.media is not None:
            print(part.media.url)
        elif part.tool_request is not None:
            print(part.tool_request.name)
        elif part.tool_response is not None:
            print(part.tool_response.output)

Part.from_text('hello') and Part(text='hello') build one. Part(root=...) raises.

Fixes & Polish

• A tool can return response(value, parts=[png]) so the model sees JSON plus media. Calling the tool yourself now returns MultipartToolResponse; out.output is the value you used to get back (#6172).
• context={'secrets': {'api_key': tenant_key}} is the Gemini key for that call. A key on config or the top-level context raises INVALID_ARGUMENT (#6236).
• ai.define_resource() is gone (#6201).
• A wrap_generate hook that swaps the model is resolved after the hook, so the swap is the model that runs (#6238).
• Dynamic action provider children show up in the registry, and those fetches are safe from more than one loop or thread (#6378, #6379).
• Google model discovery is async (#6298). The TTS fallback model no longer claims constrained output (#6301).
• pip install genkit on Python 3.9 prints how to upgrade (#6210).