Structured output

Get the agent’s final answer as typed JSON that matches your schema.

With the SDKs

Pass a Pydantic model as answer_schema or a Zod v4 schema as answerSchema. The SDK derives the JSON Schema, sends it, and parses the final answer into a typed instance. The raw wire value stays on final_changes / finalChanges. Don’t also set an agent.answer_format override: the SDK rejects the pair.

Python

from pydantic import BaseModel
from hai_agents import Client

class Story(BaseModel):
    title: str
    url: str

class Stories(BaseModel):
    stories: list[Story]

client = Client()
result = client.run_session(
    agent="h/web-surfer-flash",
    messages="Open Hacker News and list the top 3 stories with their URLs.",
    answer_schema=Stories,
)

for story in result.answer.stories:  # result.answer is a Stories instance
    print(story.title, story.url)

With a JSON Schema

Without an SDK, set a JSON Schema as answer_format on the agent, or per session with the agent.answer_format override. The settled answer (latest_answer on the session, answer on changes) is a JSON object matching it.

A real session with agent.answer_format set to a { sets: [{ name, pieces, price }] } schema:

CLI

holo run "On lego.com (US), list the 8 LEGO Star Wars sets with the most pieces, with name, piece count and price." \
  --agent h/web-surfer-flash \
  --override 'agent.answer_format={"type":"object","properties":{"sets":{"type":"array","items":{"type":"object","properties":{"name":{"type":"string"},"pieces":{"type":"integer"},"price":{"type":"number"}},"required":["name","pieces","price"]}}},"required":["sets"]}'

If the answer never matches, even after retries, the session fails with error_code: answer_validation. The SDKs raise AnswerValidationError (raw payload attached) when a completed session’s answer is missing or doesn’t parse. A failed session’s answer passes through unvalidated.

Chaining agents

A typed answer makes an agent callable like a function: one agent’s output feeds the next. Here a scout finds sources, then readers run in parallel (each counts against your concurrency limit):

Python

import asyncio
from pydantic import BaseModel
from hai_agents import AsyncClient

class Sources(BaseModel):
    urls: list[str]

class Brief(BaseModel):
    key_facts: list[str]

async def main() -> None:
    client = AsyncClient()
    scout = await client.run_session(
        agent="h/web-surfer-flash",
        messages="Find the 5 best sources on EU AI Act enforcement",
        answer_schema=Sources,
    )
    briefs = await asyncio.gather(*(
        client.run_session(agent="h/web-surfer-flash", messages=f"Extract the key facts from {url}", answer_schema=Brief)
        for url in scout.answer.urls
    ))

asyncio.run(main())