Expose a Workflow input schema

Define a Pydantic input_schema on a served workflow, then use —demo to fetch GET /workflows/ and inspect the JSON schema AgentOS exposes to clients.

with_input_schema.py

"""
Expose a Workflow input schema
==============================

Define a Pydantic input_schema on a served workflow, then use --demo to fetch
GET /workflows/{id} and inspect the JSON schema AgentOS exposes to clients.

Prerequisites: OPENAI_API_KEY is needed only for workflow runs
Run: .venvs/demo/bin/python cookbook/05_agent_os/09_serving_workflows/with_input_schema.py
Try: Run this file with --demo in another terminal
"""

import argparse
import os

import httpx
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIResponses
from agno.os import AgentOS
from agno.workflow import Step, Workflow
from pydantic import BaseModel, Field

# ---------------------------------------------------------------------------
# Create Schema-Driven Workflow
# ---------------------------------------------------------------------------

BASE_URL = os.getenv("AGENT_OS_BASE_URL", "http://localhost:7777")
WORKFLOW_ID = "research-brief-workflow"


class ResearchBrief(BaseModel):
    """Structured inputs rendered as fields by an AgentOS client."""

    topic: str = Field(description="Topic to investigate")
    focus_areas: list[str] = Field(description="Specific areas to cover")
    target_audience: str = Field(description="Who will read the research")
    sources_required: int = Field(
        default=3,
        ge=1,
        description="Minimum number of sources to consider",
    )


db = SqliteDb(
    id="workflow-input-schema-db",
    db_file="tmp/workflow_input_schema.db",
)

research_agent = Agent(
    id="schema-research-agent",
    name="Schema Research Agent",
    model=OpenAIResponses(id="gpt-5.5"),
    instructions="Follow the structured research brief and return a concise plan.",
)

research_workflow = Workflow(
    id=WORKFLOW_ID,
    name="Research Brief Workflow",
    description="Create a plan from a structured research brief.",
    db=db,
    input_schema=ResearchBrief,
    steps=[Step(name="Create Research Plan", agent=research_agent)],
)

agent_os = AgentOS(
    id="workflow-input-schema-os",
    db=db,
    workflows=[research_workflow],
)
app = agent_os.get_app()


def show_input_schema() -> None:
    """Fetch the served workflow detail and verify its JSON schema."""
    with httpx.Client(base_url=BASE_URL, timeout=30.0) as client:
        health_response = client.get("/health")
        health_response.raise_for_status()

        detail_response = client.get(f"/workflows/{WORKFLOW_ID}")
        detail_response.raise_for_status()
        detail = detail_response.json()

    input_schema = detail["input_schema"]
    properties = input_schema["properties"]
    required = set(input_schema["required"])
    expected = {"topic", "focus_areas", "target_audience", "sources_required"}

    if set(properties) != expected:
        raise RuntimeError("Workflow detail returned an unexpected input schema")
    if required != {"topic", "focus_areas", "target_audience"}:
        raise RuntimeError("Workflow detail returned unexpected required fields")

    print(f"Health: {health_response.json()['status']}")
    print(f"Workflow: {detail['id']}")
    print(f"Input schema title: {input_schema['title']}")
    print(f"Form fields: {list(properties)}")
    print(f"Required fields: {sorted(required)}")


# ---------------------------------------------------------------------------
# Run Input Schema Demo
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--demo",
        action="store_true",
        help="Fetch workflow detail from a server already running on port 7777.",
    )
    args = parser.parse_args()

    if args.demo:
        show_input_schema()
    else:
        agent_os.serve(app=app)

Run the Example

1. Set up your virtual environment

   Mac

   uv venv --python 3.12
   source .venv/bin/activate
   
2. Install dependencies

   uv pip install -U "agno[os]" openai
   
3. Export your OpenAI API key

   Mac/Linux

   export OPENAI_API_KEY="your_openai_api_key_here"
   
4. Run the example

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

   python with_input_schema.py
   

Full source: cookbook/05_agent_os/09_serving_workflows/with_input_schema.py