Stream Structured Output over AG-UI

structured_output.py

"""
Stream Structured Output over AG-UI
===================================

Give an AG-UI agent a Pydantic output schema so the streamed response follows
one predictable movie-pitch structure.

Prerequisites: OPENAI_API_KEY
Run: .venvs/demo/bin/python cookbook/05_agent_os/16_agui/structured_output.py
Try: POST "Pitch a lunar mystery" to http://localhost:7777/structured-output/agui
"""

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIResponses
from agno.os import AgentOS
from agno.os.interfaces.agui import AGUI
from pydantic import BaseModel, Field

# ---------------------------------------------------------------------------
# Create Structured Agent
# ---------------------------------------------------------------------------

class MoviePitch(BaseModel):
    title: str = Field(description="A short movie title.")
    genre: str = Field(description="The movie genre.")
    setting: str = Field(description="Where and when the story takes place.")
    characters: list[str] = Field(description="The main character names.")
    storyline: str = Field(description="A three-sentence storyline.")

db = SqliteDb(
    id="agui-structured-output-db",
    db_file="tmp/agui_structured_output.db",
)

script_writer = Agent(
    id="agui-script-writer",
    name="AG-UI Script Writer",
    model=OpenAIResponses(id="gpt-5.5"),
    db=db,
    output_schema=MoviePitch,
    instructions="Turn each request into an original, internally consistent movie pitch.",
)

agent_os = AgentOS(
    id="agui-structured-output-os",
    description="AG-UI streaming a Pydantic-validated response.",
    agents=[script_writer],
    interfaces=[AGUI(agent=script_writer, prefix="/structured-output")],
)
app = agent_os.get_app()

# ---------------------------------------------------------------------------
# Run Structured Output Server
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    agent_os.serve(app=app)

Validate the completed answer

output_schema=MoviePitch requests structured output, but parsing can fall back to text. The current AG-UI adapter can emit that text followed by RUN_FINISHED even when it does not validate as MoviePitch. Buffer the TEXT_MESSAGE_CONTENT deltas by messageId, handle RUN_ERROR, and validate the assembled answer before reading its fields.

After starting structured_output.py, save this client as consume_pitch.py beside it and run python consume_pitch.py in the same environment. It uses the same MoviePitch class, reads the server's SSE data records, checks the terminal event, and validates the assembled message:

consume_pitch.py

import json
from uuid import uuid4

import httpx
from structured_output import MoviePitch

request = {
    "threadId": str(uuid4()),
    "runId": str(uuid4()),
    "state": {},
    "messages": [
        {"id": str(uuid4()), "role": "user", "content": "Pitch a lunar mystery"}
    ],
    "tools": [],
    "context": [],
    "forwardedProps": {},
}
response = httpx.post(
    "http://localhost:7777/structured-output/agui", json=request, timeout=120
)
response.raise_for_status()
messages: dict[str, str] = {}
finished = False
for line in response.iter_lines():
    if not line.startswith("data:"):
        continue
    event = json.loads(line[5:].lstrip())
    if event["type"] == "RUN_ERROR":
        raise RuntimeError(event.get("message", "AG-UI run failed"))
    if event["type"] == "TEXT_MESSAGE_CONTENT":
        message_id = event["messageId"]
        messages[message_id] = messages.get(message_id, "") + event["delta"]
    if event["type"] == "RUN_FINISHED":
        finished = True
if not finished or len(messages) != 1:
    raise RuntimeError("Expected one complete movie-pitch message")
pitch = MoviePitch.model_validate_json(next(iter(messages.values())))
print(pitch.model_dump_json(indent=2))

This client buffers the small example response before parsing it. Invalid JSON or a schema mismatch raises an error before any movie fields are used. RUN_FINISHED describes the stream lifecycle; it is not proof of successful schema validation.

Run the Example

Set up your virtual environment

uv venv --python 3.12
source .venv/bin/activate

Install dependencies

uv pip install -U "agno[agui,os]" openai

Export your OpenAI API key

export OPENAI_API_KEY="your_openai_api_key_here"

Run the example

Save the code above as structured_output.py, then run:

python structured_output.py

Send a request

With the server running in another terminal:

Save this complete AG-UI request as agui-input.json. Replace content with the prompt for this example. Use a fresh runId for each run; keep threadId only when continuing the same session.

agui-input.json

{
  "threadId": "example-thread-1",
  "runId": "example-run-1",
  "state": {},
  "messages": [
    {"id": "example-message-1", "role": "user", "content": "Hello"}
  ],
  "tools": [],
  "context": [],
  "forwardedProps": {}
}

The endpoint accepts this object, rather than a JSON string or the REST run endpoint's message form field. The response is an SSE stream whose data values are AG-UI event objects. Inspect RUN_ERROR and RUN_FINISHED events; an HTTP 200 only establishes that the stream opened.

curl --no-buffer -H "Content-Type: application/json" --data-binary @agui-input.json http://localhost:7777/structured-output/agui

Full source: cookbook/05_agent_os/16_agui/structured_output.py