Skip to main content
By default the agent’s answer is free-form text. Set an answer_format (a JSON Schema) on the agent and it returns data that conforms to it: the answer you read from changes is then a JSON object instead of a string. Define it inline on a custom agent, or ask a catalog agent for it on a single run with overrides. The SDKs go one step further: pass a Pydantic model (Python) or Zod v4 schema (TypeScript) as answer_schema / answerSchema and the SDK derives the JSON Schema for you, then parses the final answer back into a validated, typed instance. A completed answer that does not match the schema raises AnswerValidationError with the raw payload attached; the raw wire value always stays on the result’s changes, next to the parsed answer. The schema and an agent.answer_format override are two ways to set the same field, so passing both is rejected. If the run ends without producing an answer, the parsed answer is empty (None / undefined) rather than an error. 
# `hai run` prints the structured answer once it lands
hai run "Top 3 stories on Hacker News right now?" \
  --agent h/web-surfer-flash \
  --override 'agent.answer_format={"type":"object","properties":{"stories":{"type":"array","items":{"type":"object","properties":{"title":{"type":"string"},"url":{"type":"string"}},"required":["title","url"]}}},"required":["stories"]}'

​
Agents as building blocks

With a typed answer, an agent behaves like any other function: call it, get data back, build on it. Here the stories feed a morning digest: 
digest = "\n".join(f"- {s.title} ({s.url})" for s in result.answer.stories)
send_email(to="team@yourco.com", subject="Your morning briefing", body=digest)

​
Chaining agents

A typed answer can also feed the next agent. Here one agent gathers sources and others read them in parallel: 
import asyncio
from pydantic import BaseModel
from hai_agents import AsyncClient

class Source(BaseModel):
    title: str
    url: str
    excerpt: str

class Sources(BaseModel):
    sources: list[Source]

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

async def main() -> None:
    client = AsyncClient()

    scout = await client.run_session(
        agent="h/web-surfer-flash",
        messages="Find the 5 highest-value sources on EU AI Act enforcement",
        answer_schema=Sources,
    )

    readers = await asyncio.gather(*(
        client.run_session(
            agent="h/web-surfer-flash",
            messages=f"Read this source and extract the key facts: {source.url}",
            overrides={"agent.environments[kind=web].start_url": source.url},
            answer_schema=Brief,
        )
        for source in scout.answer.sources
    ))

    for reader in readers:
        print(reader.answer.url, reader.answer.key_facts)

asyncio.run(main())
Parallel sessions count against your concurrency quota.