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")
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()
| Method | Schema Support | Streaming | Use Case |
|---|
.create() | Dict only | ✓ | Manual JSON schemas |
.parse() | Pydantic/Zod | ❌ | Type-safe non-streaming |
.stream() | Pydantic/Zod | ✓ | Type-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) => { ... },
})