Skip to main content
LLMs generate text. Applications need data structures. Structured outputs bridge this gap—define a schema (Pydantic in Python, Zod in TypeScript), and the SDK ensures responses conform with full type safety. This is essential for building reliable applications. Instead of parsing free-form text and hoping for the best, you get validated objects that your code can trust.

Client API

The client provides three methods for structured outputs:
  • .parse() - Non-streaming with type-safe schemas
  • .stream() - Streaming with type-safe schemas (context manager)
  • .create() - Dict-based schemas only

Basic Usage with .parse()

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class PersonInfo(BaseModel):
    name: str
    age: int
    occupation: str
    skills: list[str]

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-4o-mini",
        messages=[
            {"role": "user", "content": "Profile for Alice, 28, software engineer"}
        ],
        response_format=PersonInfo,
    )

    # Access parsed Pydantic model
    person = completion.choices[0].message.parsed
    print(f"{person.name}, {person.age}")
    print(f"Skills: {', '.join(person.skills)}")

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

Streaming with .stream()

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class PersonInfo(BaseModel):
    name: str
    age: int
    occupation: str
    skills: list[str]

async def main():
    client = AsyncDedalus()

    # Use context manager for streaming
    async with client.chat.completions.stream(
        model="openai/gpt-4o-mini",
        messages=[{"role": "user", "content": "Profile for Bob, 32, data scientist"}],
        response_format=PersonInfo,
    ) as stream:
        # Process events as they arrive
        async for event in stream:
            if event.type == "content.delta":
                print(event.delta, end="", flush=True)
            elif event.type == "content.done":
                # Snapshot available at content.done
                print(f"\nSnapshot: {event.parsed.name}")

        # Get final parsed result
        final = await stream.get_final_completion()
        person = final.choices[0].message.parsed
        print(f"\nFinal: {person.name}, {person.age}")

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

Optional Fields

Use Optional[T] in Python or .nullable() in Zod for nullable fields:
import asyncio
from typing import Optional
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class PartialInfo(BaseModel):
    name: str
    age: Optional[int] = None
    occupation: Optional[str] = None

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-4o-mini",
        messages=[{"role": "user", "content": "Just name: Dave"}],
        response_format=PartialInfo,
    )

    person = completion.choices[0].message.parsed
    print(f"Name: {person.name}")
    print(f"Age: {person.age or 'unknown'}")

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

Nested Models

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Skill(BaseModel):
    name: str
    years_experience: int

class DetailedProfile(BaseModel):
    name: str
    age: int
    skills: list[Skill]

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": "Profile for expert developer Alice, 28, with 5 years Python and 3 years Rust"
        }],
        response_format=DetailedProfile,
    )

    profile = completion.choices[0].message.parsed
    print(f"{profile.name}: {len(profile.skills)} skills")
    for skill in profile.skills:
        print(f"  - {skill.name}: {skill.years_experience}y")

Structured Tool Calls

Define type-safe tools with automatic argument parsing:
import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class WeatherInfo(BaseModel):
    location: str
    temperature: int
    conditions: str

async def main():
    client = AsyncDedalus()

    tools = [
        {
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "Get weather for a location",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "location": {"type": "string"}
                    },
                    "required": ["location"],
                    "additionalProperties": False,
                },
                "strict": True,
            }
        }
    ]

    completion = await client.chat.completions.parse(
        model="openai/gpt-4o-mini",
        messages=[{"role": "user", "content": "What's the weather in Paris?"}],
        tools=tools,
        response_format=WeatherInfo,
    )

    message = completion.choices[0].message
    if message.tool_calls:
        print(f"Tool called: {message.tool_calls[0].function.name}")
    elif message.parsed:
        print(f"Weather: {message.parsed.location}, {message.parsed.temperature}°C")

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

Enums and Unions

import asyncio
from typing import Literal
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class Task(BaseModel):
    title: str
    priority: Literal["low", "medium", "high", "urgent"]
    status: Literal["todo", "in_progress", "done"]
    assignee: str | None = None

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": "Create a high priority task: Fix authentication bug. Status: in progress. No assignee yet."
        }],
        response_format=Task,
    )

    task = completion.choices[0].message.parsed
    print(f"Task: {task.title}")
    print(f"Priority: {task.priority}, Status: {task.status}")

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

DedalusRunner API

The Runner supports response_format with automatic schema conversion:
import asyncio
from dedalus_labs import AsyncDedalus, DedalusRunner
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class WeatherResponse(BaseModel):
    location: str
    temperature: int
    summary: str

async def get_weather(location: str) -> str:
    """Get weather for a location."""
    return f"Sunny, 72°F in {location}"

async def main():
    client = AsyncDedalus()
    runner = DedalusRunner(client)

    result = await runner.run(
        input="What's the weather in Paris?",
        model="openai/gpt-4o-mini",
        tools=[get_weather],
        response_format=WeatherResponse,
        max_steps=5,
    )

    print(result.final_output)

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

.create() vs .parse() vs .stream()

MethodSchema SupportStreamingUse Case
.create()Dict onlyManual JSON schemas
.parse()Pydantic/ZodType-safe non-streaming
.stream()Pydantic/ZodType-safe streaming
.create() will throw a TypeError if you pass a Pydantic/Zod model directly. Use .parse() or .stream() for type-safe schemas.

Error Handling

import asyncio
from dedalus_labs import AsyncDedalus
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

class PersonInfo(BaseModel):
    name: str
    age: int

async def main():
    client = AsyncDedalus()

    completion = await client.chat.completions.parse(
        model="openai/gpt-4o-mini",
        messages=[{"role": "user", "content": "Generate harmful content"}],
        response_format=PersonInfo,
    )

    message = completion.choices[0].message
    if message.refusal:
        print(f"Model refused: {message.refusal}")
    elif message.parsed:
        print(f"Parsed: {message.parsed.name}")
    else:
        print("No response or parsing failed")

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

Supported Models

The SDK’s .parse() and .stream() methods work across all providers. Schema enforcement varies: Strict Enforcement (CFG-based, schema guarantees):
  • openai/* - Context-free grammar compilation
  • xai/* - Native schema validation
  • fireworks_ai/* - Native schema validation (select models)
  • deepseek/* - Native schema validation (select models)
Best-Effort (schema sent for guidance, no guarantees):
  • 🟡 google/* - Schema forwarded to generationConfig.responseSchema
  • 🟡 anthropic/* - Prompt-based JSON generation (~85-90% success rate)
For google/* and anthropic/* models, always validate parsed output and implement retry logic.

Provider Examples

Same code, different models. Swap the model string and everything else stays the same.

Python

from dedalus_labs import AsyncDedalus
from pydantic import BaseModel

class PersonInfo(BaseModel):
    name: str
    age: int
    occupation: str

client = AsyncDedalus()
result = await client.chat.completions.parse(
    model="openai/gpt-4o-mini",
    messages=[{"role": "user", "content": "Profile for Alice, 28, engineer"}],
    response_format=PersonInfo,
)
print(result.choices[0].message.parsed)

TypeScript

import Dedalus from 'dedalus-labs';
import { zodResponseFormat } from 'dedalus-labs/helpers/zod';
import { z } from 'zod';

const PersonInfo = z.object({
  name: z.string(),
  age: z.number(),
  occupation: z.string(),
});

const client = new Dedalus();
const result = await client.chat.completions.parse({
  model: 'openai/gpt-4o-mini',
  messages: [{ role: 'user', content: 'Profile for Alice, 28, engineer' }],
  response_format: zodResponseFormat(PersonInfo, 'person'),
});
console.log(result.choices[0]?.message.parsed);

Quick Reference

Python (Pydantic)

from dedalus_labs import AsyncDedalus
from pydantic import BaseModel

class MyModel(BaseModel):
    field: str

client = AsyncDedalus()
result = await client.chat.completions.parse(
    model="openai/gpt-4o-mini",
    messages=[...],
    response_format=MyModel,
)
parsed = result.choices[0].message.parsed

TypeScript (Zod)

import Dedalus from 'dedalus-labs';
import { zodResponseFormat } from 'dedalus-labs/helpers/zod';
import { z } from 'zod';

const MySchema = z.object({ field: z.string() });

const client = new Dedalus();
const result = await client.chat.completions.parse({
  model: 'openai/gpt-4o-mini',
  messages: [...],
  response_format: zodResponseFormat(MySchema, 'my_schema'),
});
const parsed = result.choices[0]?.message.parsed;

Zod Helpers

import { zodResponseFormat, zodFunction } from 'dedalus-labs/helpers/zod';

// For response schemas
zodResponseFormat(MyZodSchema, 'schema_name')

// For tool definitions
zodFunction({
  name: 'tool_name',
  description: 'What the tool does',
  parameters: z.object({ ... }),
  function: (args) => { ... },
})