Use Claude Structured Output and Streaming

Claude structured output constrains the final agent result to a JSON Schema, while streaming exposes work and partial output as the agent loop progresses. Consume the final structured field only after a successful result; treat deltas as provisional UI events. Validate business rules after schema conformance and design reconnect/cancellation behavior explicitly.

What will you be able to do?

You will configure a triage schema, handle the final result, and choose between single prompt, streaming input, and partial output modes.

How do you request a schema?

python

import asyncio

from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query


SCHEMA = {
    "type": "object",
    "properties": {
        "category": {"enum": ["billing", "technical", "general"]},
        "summary": {"type": "string", "maxLength": 300},
        "needs_human": {"type": "boolean"},
    },
    "required": ["category", "summary", "needs_human"],
    "additionalProperties": False,
}


async def main() -> None:
    options = ClaudeAgentOptions(
        max_turns=3,
        output_format={"type": "json_schema", "schema": SCHEMA},
    )
    async for message in query(
        prompt="Classify: I was charged twice.",
        options=options,
    ):
        if isinstance(message, ResultMessage):
            if message.subtype != "success":
                raise RuntimeError(message.subtype)
            print(message.structured_output)


if __name__ == "__main__":
    asyncio.run(main())

Check the current reference for exact result fields in your installed SDK. After receiving the object, validate domain rules such as “billing actions require human review.”

How do streaming modes differ?

A string prompt is simplest for one-shot input.
Streaming input lets your application send messages over time and supports interactive control patterns.
Partial output messages expose raw response deltas for responsive interfaces.
Assistant messages expose completed turns and tool-call blocks.
The result message provides authoritative disposition, usage, and final data.

Do not parse JSON incrementally from text deltas when the SDK supplies final structured output.

How do you prove it works?

Test success, schema mismatch, max-turn stop, consumer cancellation, slow consumer/backpressure, and reconnect. Store normalized events with sequence numbers; store final structured outcome separately.

Failure injection: Close the client after it displays 90% of a streamed answer. The run is not automatically complete. Reconnect to persisted events or report that observation detached while the worker followed its cancellation policy.

What mistakes should you avoid?

Treating streamed text as committed state.
Skipping semantic validation because JSON is valid.
Breaking iteration before the SDK can clean up.
Letting one slow client block the agent worker.
Exposing raw provider events as a permanent public protocol.

Check your understanding

When is structured output authoritative?
What is the difference between an assistant message and a raw stream event?
Which layer validates business rules?

Expert extension

Build an event adapter with replay from Last-Event-ID, a final-outcome endpoint, and tests for duplicate, missing, and out-of-order delivery.

Official references

Use Claude Structured Output and Streaming

How do you request a schema?

python

import asyncio

from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query


SCHEMA = {
    "type": "object",
    "properties": {
        "category": {"enum": ["billing", "technical", "general"]},
        "summary": {"type": "string", "maxLength": 300},
        "needs_human": {"type": "boolean"},
    },
    "required": ["category", "summary", "needs_human"],
    "additionalProperties": False,
}


async def main() -> None:
    options = ClaudeAgentOptions(
        max_turns=3,
        output_format={"type": "json_schema", "schema": SCHEMA},
    )
    async for message in query(
        prompt="Classify: I was charged twice.",
        options=options,
    ):
        if isinstance(message, ResultMessage):
            if message.subtype != "success":
                raise RuntimeError(message.subtype)
            print(message.structured_output)


if __name__ == "__main__":
    asyncio.run(main())

Check the current reference for exact result fields in your installed SDK. After receiving the object, validate domain rules such as “billing actions require human review.”

How do streaming modes differ?

A string prompt is simplest for one-shot input.

Streaming input lets your application send messages over time and supports interactive control patterns.

Partial output messages expose raw response deltas for responsive interfaces.

Assistant messages expose completed turns and tool-call blocks.

The result message provides authoritative disposition, usage, and final data.

Do not parse JSON incrementally from text deltas when the SDK supplies final structured output.

How do you prove it works?

Failure injection: Close the client after it displays 90% of a streamed answer. The run is not automatically complete. Reconnect to persisted events or report that observation detached while the worker followed its cancellation policy.