Structured Output (Response Schema)¶

Some calls do not want prose back. A triage step wants a department and an urgency flag, a judge wants a verdict, a classifier wants a label. Asking for JSON in the prompt and parsing whatever comes back works most of the time, and fails on the answer that wraps the object in a sentence or stops halfway through it.

AIContext.response_schema removes that guesswork. It carries a JSON Schema, the provider constrains its output natively, and generate() returns one JSON document that satisfies the schema. RoomKit checks the document against the schema before handing it back, whatever the server claims to have done. If there is no such document, generate() raises an error that says why. It never returns prose in its place.

Quick start¶

import json

from roomkit import ResponseSchemaError
from roomkit.providers.ai import AIContext, AIMessage
from roomkit.providers.gemini import GeminiAIProvider, GeminiConfig

TRIAGE = {
    "type": "object",
    "properties": {
        "department": {"type": "string", "enum": ["billing", "technical", "sales", "other"]},
        "urgent": {"type": "boolean"},
        "summary": {"type": "string", "description": "One short sentence."},
    },
    "required": ["department", "urgent", "summary"],
    "additionalProperties": False,
}

provider = GeminiAIProvider(GeminiConfig(api_key="..."))
context = AIContext(
    system_prompt="You triage customer support messages.",
    messages=[AIMessage(role="user", content="I was charged twice for March.")],
    response_schema=TRIAGE,
)

try:
    response = await provider.generate(context)
except ResponseSchemaError as exc:
    print(f"no triage: {exc.reason}")
else:
    triage = json.loads(response.content)

response.content stays a string. Turning it into a typed value (with json.loads, or a Pydantic model's model_validate_json) is the caller's work.

A runnable version is in examples/ai_response_schema.py. It runs offline on the mock, or against Gemini, Anthropic, OpenAI or a local Ollama.

The portable subset¶

Every provider reads its own dialect of JSON Schema. OpenAI's strict mode wants every property required; Anthropic refuses numeric bounds; Gemini reads its own subset. RoomKit accepts only what they all take, so a schema written once runs on any provider that supports response schemas:

Keyword	Rule
`type`	`object`, `array`, `string`, `number`, `integer` or `boolean`, as a single string. The root is an `object`.
`properties`	Required on an object, possibly empty.
`required`	Required on an object, and lists every property, each once.
`additionalProperties`	Required on an object, and `false`.
`items`	Required on an array.
`enum`	On a string only: a non-empty list of distinct strings.
`title`, `description`	Allowed anywhere, as strings. The model reads them.

Nothing else passes: no null or optional fields, no anyOf, no $ref, no minimum or maxLength. For an answer that may be absent, use a required field with an agreed empty value ("", []).

The check runs when the context is built, when the field is assigned, and through model_copy(update=...). A schema outside the subset fails there, before any provider is called. check_portable_schema(schema) runs the same check on its own, for a schema you build at startup.

Providers¶

Provider	Native field	Supported
OpenAI, Azure, xAI, Cerebras, OpenRouter, LiteLLM, vLLM, llama.cpp	strict `json_schema` `response_format`	yes
DeepSeek, Qwen	—	no: only free-form JSON mode is documented
Anthropic	`output_config.format`	yes
Gemini, Gemini Vertex	`response_mime_type` + `response_json_schema`	yes
Mistral	strict `json_schema` `response_format`	yes
Ollama	`format`	yes
PolarGrid	`json_schema` `response_format`	yes

On OpenRouter, a constrained request also asks for provider.require_parameters, so it is only routed to an upstream that honours the format. Routing preferences set in extra_body are kept.

provider.supports_response_schema gives the provider's answer. It is a default, not a promise for every model: a model older than the feature may still reject the request, which surfaces as a ProviderError.

Whether an OpenAI-compatible server applies the format depends on the server, so OpenAIConfig, AzureAIConfig and VLLMConfig take supports_response_schema= to state it:

OpenAIConfig(
    api_key="...",
    base_url="http://my-server:8000/v1",
    model="my-model",
    supports_response_schema=False,  # this server ignores response_format
)

When there is no document¶

ResponseSchemaError is a ProviderError that is never retryable: the same request fails the same way. Its reason tells the cases apart:

`reason`	Meaning
`unsupported`	The call cannot carry a schema: the provider does not support one, or cannot support one beside the turn's `tools`. Raised before any request is sent.
`refusal`	The model declined to answer: a refusal field (OpenAI), a `refusal` stop reason (Anthropic), a safety stop or a blocked prompt (Gemini), a content filter.
`truncated`	The answer was cut: the output cap, or the context window filling up. Raise `max_tokens`; a reasoning model spends part of it thinking.
`invalid_json`	The text is not a JSON document satisfying the schema: not JSON at all, or JSON of another shape. That happens with a server that accepted the constraint and did not apply it.

Running on any provider¶

A component that must work with every provider checks the property first, and where it is false asks for JSON in the prompt and checks the answer itself. schema_mismatch is the same check the providers run:

from roomkit.providers.ai import schema_mismatch

if provider.supports_response_schema:
    context = context.model_copy(update={"response_schema": TRIAGE})
    triage = json.loads((await provider.generate(context)).content)
else:
    instruction = "Answer with one JSON object matching: " + json.dumps(TRIAGE)
    context.system_prompt = f"{context.system_prompt or ''}\n{instruction}".strip()
    triage = json.loads((await provider.generate(context)).content)  # may raise
    if (problem := schema_mismatch(TRIAGE, triage)) is not None:
        raise ValueError(problem)

Streaming¶

generate_stream() and generate_structured_stream() carry the schema too. The text deltas are the JSON as the model writes it, partial and provisional; the whole document is checked before the done event, and the error takes the done event's place when the check fails. So a consumer shows the text as it streams, if it likes, and acts on it only once StreamDone arrives:

async for event in provider.generate_structured_stream(context):
    if isinstance(event, StreamTextDelta):
        buffer.append(event.text)          # partial JSON, provisional
    elif isinstance(event, StreamDone):
        triage = json.loads("".join(buffer))  # checked against the schema

Tools in the same turn¶

A turn may carry tools and a schema together where provider.supports_response_schema_with_tools is true: OpenAI's own endpoint and Azure, Anthropic, Gemini. The model either calls tools or answers in the schema. A response carrying tool calls is a step of the loop and is not checked; the final answer, the one without tool calls, is.

Where the constraint is a decoding grammar (Ollama, vLLM, llama.cpp, PolarGrid) the pair is refused before any request. The grammar forces every token into the schema, so the model cannot call the tool: asked for the weather with a get_weather tool and a schema, a local Ollama model answered a schema-valid temperature it had made up. supports_response_schema_with_tools=True on OpenAIConfig, AzureAIConfig or VLLMConfig states otherwise for a server you know combines them.

AIChannel turns¶

An AIChannel answers in a schema the same way. Set it for every turn, per turn from a config provider, or per room in the binding metadata; the binding wins, then the config provider, then the channel:

triage = AIChannel("triage", provider=provider, response_schema=TRIAGE)

# One room answers in another shape
await kit.attach_channel(
    "vip-room", "triage", category=ChannelCategory.INTELLIGENCE,
    metadata={"response_schema": URGENCY},
)

The turn's answer is then one JSON document, checked against the schema before any of it reaches the room. A streamed answer is held until the check passes and arrives in one piece; one that fails the check is never delivered nor stored, and the turn fails through ON_ERROR.

What fails the turn:

Case	When	Reason
The provider cannot honour a schema	before any request	`unsupported`
The turn has tools and the provider cannot combine them with a schema	before any request	`unsupported`
The model refuses, the output cap or the context window cuts the answer, or the text is not a document satisfying the schema	after the answer	`refusal`, `truncated`, `invalid_json`
The tool loop stops before a final answer (`max_tool_rounds`, the loop timeout, a turn budget, a stop)	when the loop ends	`truncated`

The turn's tools include the ones the channel adds itself: skills, the sandbox, planning (enable_planning=True), and orchestration tools such as submit_result. A channel with any of them needs a provider that combines tools and a schema.

When the model calls tools, a round can carry text of its own ("Let me look that up."). That text reaches the room as its own message, as in any turn; the turn's last text message is the JSON document.

A channel default outside the portable subset fails when the channel is built.

Vision¶

A VisionProvider answers a frame in a schema the same way: analyze_frame(frame, prompt=..., response_schema=SCHEMA) returns a VisionResult whose description is the JSON document, checked before it is returned. GeminiVisionProvider and OpenAIVisionProvider support it (OpenAIVisionConfig(supports_response_schema=False) for a server that does not); vision.supports_response_schema says which do. The screen input tools' click_element uses it to locate an element.

The rules live in RFC §6.7.

Testing¶

MockAIProvider(response_schema=True) follows the same contract. The scripted answer must be JSON, and an AIResponse scripted with finish_reason="refusal" or "length" raises the matching error:

from roomkit.providers.ai import AIResponse, MockAIProvider

provider = MockAIProvider(
    ai_responses=[AIResponse(content="", finish_reason="refusal")],
    response_schema=True,
)