@@ -6,7 +6,7 @@
 
 import ai
 
-model = ai.ai_gateway("anthropic/claude-sonnet-4")
+model = ai.ai_gateway("anthropic/claude-sonnet-4.6")
 
 
 class Recipe(pydantic.BaseModel):
@@ -20,9 +20,21 @@ class Recipe(pydantic.BaseModel):
 
 
 async def main() -> None:
-    # Broken for now: stream(output_type=...) requests JSON/schema mode, but
-    # the stream wrapper does not yet validate final text into s.output.
-    raise RuntimeError("structured output aggregation needs to be implemented")
+    # Stream with structured output — watch JSON arrive, get validated at the end.
+    async with ai.stream(model, messages, output_type=Recipe) as s:
+        async for event in s:
+            if isinstance(event, ai.events.TextDelta):
+                print(event.chunk, end="", flush=True)
+            elif isinstance(event, ai.events.StreamEnd):
+                print()
+
+    # After iteration, s.output validates the streamed JSON against
+    # the output_type the stream was opened with.
+    recipe = s.output
+    assert isinstance(recipe, Recipe)
+    print(f"\n\nParsed recipe: {recipe.name}")
+    print(f"  Ingredients: {', '.join(recipe.ingredients)}")
+    print(f"  Prep time: {recipe.prep_time_minutes} min")
 
 
 if __name__ == "__main__":
 
@@ -84,6 +84,7 @@ def __init__(
         gen: AsyncGenerator[types.events.Event],
         *,
         seed_message: types.messages.Message | None = None,
+        output_type: type[pydantic.BaseModel] | None = None,
     ) -> None:
         """Wrap an event generator.
 
@@ -93,12 +94,18 @@ def __init__(
         being rebuilt from synthetic events.  When ``None`` (default),
         an empty assistant message is created and rebuilt from the
         incoming events.
+
+        ``output_type`` is the Pydantic model the request was constrained
+        to.  When set, ``Stream.output`` validates the streamed JSON text
+        against it.  When ``None`` (default), ``Stream.output`` returns
+        the concatenated text content unchanged.
         """
         self._gen = gen
         self._message: types.messages.Message = seed_message or types.messages.Message(
             role="assistant", parts=[]
         )
         self._parts: dict[str, types.messages.Part] = {}
+        self._output_type = output_type
 
     async def aclose(self) -> None:
         await self._gen.aclose()
@@ -146,7 +153,15 @@ def tool_calls(self) -> list[types.messages.ToolCallPart]:
 
     @property
     def output(self) -> Any:
-        return self._message.output
+        """Return the streamed output as the ``output_type`` passed in.
+
+        Defaults to the concatenated message text.  When a Pydantic
+        model subclass was passed, validates the streamed JSON against
+        it and returns the parsed instance.
+        """
+        if self._output_type is None:
+            return self._message.text
+        return self._output_type.model_validate_json(self._message.text)
 
     def _aggregate_event(self, event: types.events.Event) -> dict[str, Any]:
         updates: dict[str, Any] = {}
@@ -393,7 +408,11 @@ async def _stream(
 ) -> AsyncIterator[Stream]:
     if messages and messages[-1].replay:
         last = messages[-1]
-        s = Stream(_replay_tool_calls(last), seed_message=last.model_copy(deep=True))
+        s = Stream(
+            _replay_tool_calls(last),
+            seed_message=last.model_copy(deep=True),
+            output_type=output_type,
+        )
     else:
         prepared = integrity.prepare_messages(messages)
         request = StreamRequest(
@@ -403,7 +422,7 @@ async def _stream(
             output_type,
             params,
         )
-        s = Stream(executor._do_stream(request))
+        s = Stream(executor._do_stream(request), output_type=output_type)
     try:
         yield s
     finally:
 
@@ -18,7 +18,6 @@
     Message,
     Part,
     ReasoningPart,
-    StructuredOutputPart,
     TextPart,
     ToolCallPart,
     ToolResultPart,
@@ -30,7 +29,6 @@
     ToolResultPart,
     ReasoningPart,
     HookPart,
-    StructuredOutputPart,
     FilePart,
 )
 
 
@@ -206,8 +206,8 @@ async def replay_message_events(
     not recoverable from a complete message.
 
     Parts with no model-layer event analog — ``ToolResultPart``,
-    ``HookPart``, ``StructuredOutputPart`` — are skipped silently; they
-    are agent-layer concerns and never appear on the model stream.
+    ``HookPart`` — are skipped silently; they are agent-layer concerns
+    and never appear on the model stream.
     """
     yield StreamStart()
     for part in msg.parts:
 
@@ -1,4 +1,3 @@
-import importlib
 import uuid
 from typing import Annotated, Any, Literal, Self
 
@@ -110,56 +109,6 @@ class HookPart[T](pydantic.BaseModel):
     model_config = pydantic.ConfigDict(frozen=True)
 
 
-# todo redo this structured output situation and simplify it
-def _resolve_class(fully_qualified_name: str) -> type[pydantic.BaseModel]:
-    """Import and return a class from its fully qualified name.
-
-    E.g. ``"myapp.models.WeatherForecast"`` → the ``WeatherForecast`` class.
-    """
-    module_path, _, class_name = fully_qualified_name.rpartition(".")
-    if not module_path:
-        raise ImportError(
-            f"Cannot resolve '{fully_qualified_name}': "
-            "expected a fully qualified name like 'mypackage.module.ClassName'"
-        )
-    module = importlib.import_module(module_path)
-    cls = getattr(module, class_name, None)
-    if cls is None:
-        raise ImportError(f"Module '{module_path}' has no attribute '{class_name}'")
-    if not (isinstance(cls, type) and issubclass(cls, pydantic.BaseModel)):
-        raise TypeError(
-            f"'{fully_qualified_name}' is not a pydantic.BaseModel subclass"
-        )
-    return cls
-
-
-class StructuredOutputPart(pydantic.BaseModel):
-    """Part containing a validated structured output from the LLM.
-
-    ``data`` stores the parsed JSON dict (always serializable).
-    ``output_type_name`` stores the fully qualified class name so the typed
-    Pydantic model can be lazily rehydrated via the ``value`` property.
-    """
-
-    model_config = pydantic.ConfigDict(frozen=True)
-
-    id: str = pydantic.Field(default_factory=generate_id)
-    data: dict[str, Any]
-    output_type_name: str
-    kind: Literal["structured_output"] = "structured_output"
-    provider_metadata: dict[str, Any] | None = None
-
-    _hydrated: Any = pydantic.PrivateAttr(default=None)
-
-    @property
-    def value(self) -> Any:
-        """Lazily resolve the output type class and validate ``data`` into it."""
-        if self._hydrated is None:
-            cls = _resolve_class(self.output_type_name)
-            self._hydrated = cls.model_validate(self.data)
-        return self._hydrated
-
-
 class FilePart(pydantic.BaseModel):
     """File, image, or audio content part.
 
@@ -229,7 +178,6 @@ def from_bytes(
     | BuiltinToolReturnPart
     | ReasoningPart
     | HookPart[Any]
-    | StructuredOutputPart
     | FilePart,
     pydantic.Field(discriminator="kind"),
 ]
@@ -288,11 +236,3 @@ def images(self) -> list[FilePart]:
     @property
     def videos(self) -> list[FilePart]:
         return [p for p in self.files if p.media_type.startswith("video/")]
-
-    @property
-    def output(self) -> Any:
-        """Parsed structured output from the first structured-output part."""
-        for part in self.parts:
-            if isinstance(part, StructuredOutputPart):
-                return part.value
-        return None
@@ -139,8 +139,6 @@ async def emit_events_for_messages(
                     media_type=part.media_type,
                     data=part.data if isinstance(part.data, str) else "",
                 )
-            # StructuredOutputPart is not a streamed part; tests that need it
-            # construct a tailored adapter directly.
     yield events_.StreamEnd(usage=usage)
 
 
 
@@ -120,7 +120,7 @@ def test_internal_strict_raises() -> None:
 
 
 # ---------------------------------------------------------------------------
-# Internal parts (HookPart, StructuredOutputPart)
+# Internal parts (HookPart)
 # ---------------------------------------------------------------------------
 
 
 
@@ -2,70 +2,11 @@
 
 from __future__ import annotations
 
-import pydantic
 import pytest
 
 from ai.types import messages, usage
 
 
-class _Weather(pydantic.BaseModel):
-    city: str
-    temperature: float
-
-
-_WEATHER_DATA = {"city": "SF", "temperature": 62.0}
-_WEATHER_TYPE_NAME = f"{_Weather.__module__}.{_Weather.__qualname__}"
-
-
-def test_structured_output_part_value() -> None:
-    part = messages.StructuredOutputPart(
-        data=_WEATHER_DATA, output_type_name=_WEATHER_TYPE_NAME
-    )
-    val = part.value
-    assert isinstance(val, _Weather)
-    assert val.city == "SF"
-    assert part.value is val
-
-
-def test_structured_output_part_bad_class_name() -> None:
-    part = messages.StructuredOutputPart(
-        data=_WEATHER_DATA, output_type_name="nonexistent.module.Cls"
-    )
-    with pytest.raises(ImportError):
-        _ = part.value
-
-
-def test_message_output_from_part() -> None:
-    m = messages.Message(
-        id="m1",
-        role="assistant",
-        parts=[
-            messages.TextPart(text="{}"),
-            messages.StructuredOutputPart(
-                data=_WEATHER_DATA, output_type_name=_WEATHER_TYPE_NAME
-            ),
-        ],
-    )
-    assert isinstance(m.output, _Weather)
-    assert m.output.city == "SF"
-
-
-def test_structured_output_round_trip() -> None:
-    m = messages.Message(
-        id="m1",
-        role="assistant",
-        parts=[
-            messages.TextPart(text="{}"),
-            messages.StructuredOutputPart(
-                data=_WEATHER_DATA, output_type_name=_WEATHER_TYPE_NAME
-            ),
-        ],
-    )
-    restored = messages.Message.model_validate(m.model_dump())
-    assert isinstance(restored.output, _Weather)
-    assert restored.output.city == "SF"
-
-
 def test_usage_add_merges_optional_fields() -> None:
     a = usage.Usage(
         input_tokens=100,
Original file line number	Diff line number	Diff line change
@@ -139,8 +139,6 @@ async def emit_events_for_messages(
139	139	media_type=part.media_type,
140	140	data=part.data if isinstance(part.data, str) else "",
141	141	)
142		- # StructuredOutputPart is not a streamed part; tests that need it
143		- # construct a tailored adapter directly.
144	142	yield events_.StreamEnd(usage=usage)
145	143
146	144