Implement structured output support delegated to providers

Author: anbuzin

examples/samples/structured_output.py

@@ -0,0 +1,48 @@
+import asyncio
+import os
+
+import pydantic
+
+import vercel_ai_sdk as ai
+
+
+class WeatherForecast(pydantic.BaseModel):
+    city: str
+    temperature: float
+    conditions: str
+    humidity: int
+    wind_speed: float
+
+
+async def main() -> None:
+    llm = ai.openai.OpenAIModel(
+        model="anthropic/claude-sonnet-4",
+        base_url="https://ai-gateway.vercel.sh/v1",
+        api_key=os.environ.get("AI_GATEWAY_API_KEY"),
+    )
+
+    messages = ai.make_messages(
+        system="You are a weather assistant. Respond with realistic weather data.",
+        user="What's the weather like in San Francisco right now?",
+    )
+
+    # Streaming: watch the JSON arrive incrementally, get validated output at the end
+    print("--- Streaming ---")
+    async for msg in llm.stream(messages, output_type=WeatherForecast):
+        if msg.text_delta:
+            print(msg.text_delta, end="", flush=True)
+        if msg.output:
+            print(f"\n\nParsed: {msg.output}")
+
+    # Non-streaming: get the validated output directly
+    print("\n--- Buffer ---")
+    msg = await llm.buffer(messages, output_type=WeatherForecast)
+    print(f"City: {msg.output.city}")
+    print(f"Temperature: {msg.output.temperature}")
+    print(f"Conditions: {msg.output.conditions}")
+    print(f"Humidity: {msg.output.humidity}%")
+    print(f"Wind: {msg.output.wind_speed} mph")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/vercel_ai_sdk/anthropic/__init__.py

@@ -6,6 +6,7 @@
 from typing import Any, override
 
 import anthropic
+import pydantic
 
 from .. import core
 
@@ -119,6 +120,7 @@ async def stream_events(
         self,
         messages: list[core.messages.Message],
         tools: Sequence[core.tools.ToolLike] | None = None,
+        output_type: type[pydantic.BaseModel] | None = None,
     ) -> AsyncGenerator[core.llm.StreamEvent]:
         """Yield raw stream events from Anthropic API."""
         system_prompt, anthropic_messages = _messages_to_anthropic(messages)
@@ -140,12 +142,30 @@ async def stream_events(
                 "budget_tokens": self._budget_tokens,
             }
 
+        # Structured output: use beta API with output_format
+        use_beta = False
+        if output_type is not None:
+            from anthropic.lib._parse._transform import transform_schema
+
+            kwargs["output_format"] = {
+                "type": "json_schema",
+                "schema": transform_schema(output_type),
+            }
+            kwargs["betas"] = ["structured-outputs-2025-11-13"]
+            use_beta = True
+
         # Track block types by index to know what End event to emit
         block_types: dict[int, str] = {}  # index -> "text" | "thinking" | "tool_use"
         tool_ids: dict[int, str] = {}  # index -> tool_call_id
         signature_buffer: dict[int, str] = {}  # index -> accumulated signature
 
-        async with self._client.messages.stream(**kwargs) as stream:
+        stream_cm: Any  # BetaAsyncMessageStreamManager | AsyncMessageStreamManager
+        if use_beta:
+            stream_cm = self._client.beta.messages.stream(**kwargs)
+        else:
+            stream_cm = self._client.messages.stream(**kwargs)
+
+        async with stream_cm as stream:
             async for event in stream:
                 if event.type == "content_block_start":
                     block = event.content_block
@@ -208,8 +228,17 @@ async def stream(
         self,
         messages: list[core.messages.Message],
         tools: Sequence[core.tools.ToolLike] | None = None,
+        output_type: type[pydantic.BaseModel] | None = None,
     ) -> AsyncGenerator[core.messages.Message]:
-        """Stream Messages (uses StreamProcessor internally)."""
+        """Stream Messages (uses StreamHandler internally)."""
         handler = core.llm.StreamHandler()
-        async for event in self.stream_events(messages, tools):
-            yield handler.handle_event(event)
+        msg: core.messages.Message | None = None
+        async for event in self.stream_events(messages, tools, output_type=output_type):
+            msg = handler.handle_event(event)
+            yield msg
+
+        # After stream completes, validate and yield final message with output
+        if output_type is not None and msg is not None and msg.text:
+            msg = msg.model_copy()
+            msg.output = output_type.model_validate_json(msg.text)
+            yield msg

src/vercel_ai_sdk/core/llm.py

@@ -4,6 +4,8 @@
 import dataclasses
 from collections.abc import AsyncGenerator, Sequence
 
+import pydantic
+
 from . import messages as messages_
 from . import tools as tools_
 
@@ -216,6 +218,7 @@ async def stream(
         self,
         messages: list[messages_.Message],
         tools: Sequence[tools_.ToolLike] | None = None,
+        output_type: type[pydantic.BaseModel] | None = None,
     ) -> AsyncGenerator[messages_.Message]:
         raise NotImplementedError
         yield
@@ -224,10 +227,11 @@ async def buffer(
         self,
         messages: list[messages_.Message],
         tools: Sequence[tools_.ToolLike] | None = None,
+        output_type: type[pydantic.BaseModel] | None = None,
     ) -> messages_.Message:
         """Drain the stream and return the final message."""
         final = None
-        async for msg in self.stream(messages, tools):
+        async for msg in self.stream(messages, tools, output_type=output_type):
             final = msg
         if final is None:
             raise ValueError("LLM produced no messages")

src/vercel_ai_sdk/core/messages.py

@@ -84,6 +84,7 @@ class Message(pydantic.BaseModel):
     parts: list[Part]
     id: str = pydantic.Field(default_factory=_gen_id)
     label: str | None = None
+    output: Any = pydantic.Field(default=None, exclude=True)
 
     @property
     def is_done(self) -> bool:

src/vercel_ai_sdk/core/streams.py

@@ -30,6 +30,13 @@ def text(self) -> str:
             return self.last_message.text
         return ""
 
+    @property
+    def output(self) -> Any:
+        """Parsed structured output from the last message, if available."""
+        if self.last_message:
+            return self.last_message.output
+        return None
+
 
 Stream = Callable[[], AsyncGenerator[messages_.Message]]
 # maybe it should have a name and an id inferred from LLM outputs

src/vercel_ai_sdk/openai/__init__.py

@@ -5,6 +5,7 @@
 from typing import Any, override
 
 import openai
+import pydantic
 
 from .. import core
 
@@ -138,6 +139,7 @@ async def stream_events(
         self,
         messages: list[core.messages.Message],
         tools: Sequence[core.tools.ToolLike] | None = None,
+        output_type: type[pydantic.BaseModel] | None = None,
     ) -> AsyncGenerator[core.llm.StreamEvent]:
         """Yield raw stream events from OpenAI API."""
         openai_messages = _messages_to_openai(messages)
@@ -151,6 +153,18 @@ async def stream_events(
         if openai_tools:
             kwargs["tools"] = openai_tools
 
+        if output_type is not None:
+            from openai.lib._pydantic import to_strict_json_schema
+
+            kwargs["response_format"] = {
+                "type": "json_schema",
+                "json_schema": {
+                    "name": output_type.__name__,
+                    "schema": to_strict_json_schema(output_type),
+                    "strict": True,
+                },
+            }
+
         # Enable reasoning/thinking via Vercel AI Gateway's unified format
         # See: https://vercel.com/docs/ai-gateway/openai-compat/advanced
         if self._thinking:
@@ -249,8 +263,17 @@ async def stream(
         self,
         messages: list[core.messages.Message],
         tools: Sequence[core.tools.ToolLike] | None = None,
+        output_type: type[pydantic.BaseModel] | None = None,
     ) -> AsyncGenerator[core.messages.Message]:
         """Stream Messages (uses StreamHandler internally)."""
         handler = core.llm.StreamHandler()
-        async for event in self.stream_events(messages, tools):
-            yield handler.handle_event(event)
+        msg: core.messages.Message | None = None
+        async for event in self.stream_events(messages, tools, output_type=output_type):
+            msg = handler.handle_event(event)
+            yield msg
+
+        # After stream completes, validate and yield final message with output
+        if output_type is not None and msg is not None and msg.text:
+            msg = msg.model_copy()
+            msg.output = output_type.model_validate_json(msg.text)
+            yield msg

tests/conftest.py

@@ -2,6 +2,8 @@
 
 from collections.abc import AsyncGenerator, Sequence
 
+import pydantic
+
 import vercel_ai_sdk as ai
 from vercel_ai_sdk.core import messages
 
@@ -18,15 +20,23 @@ async def stream(
         self,
         messages: list[messages.Message],
         tools: Sequence[ai.ToolLike] | None = None,
+        output_type: type[pydantic.BaseModel] | None = None,
     ) -> AsyncGenerator[messages.Message]:
         if self._call_index >= len(self._responses):
             raise RuntimeError("MockLLM: no more responses configured")
         self.call_count += 1
         seq = self._responses[self._call_index]
         self._call_index += 1
+        msg = None
         for msg in seq:
             yield msg
 
+        # Simulate structured output validation (matching real provider behavior)
+        if output_type is not None and msg is not None and msg.text:
+            msg = msg.model_copy()
+            msg.output = output_type.model_validate_json(msg.text)
+            yield msg
+
 
 def text_msg(
     text: str, *, id: str = "msg-1", state: str = "done", delta: str | None = None