> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agno.com/llms.txt
> Use this file to discover all available pages before exploring further.

# 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.

```python with_input_schema.py theme={null}
"""
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

<Steps>
  <Snippet file="create-venv-step.mdx" />

  <Step title="Install dependencies">
    ```bash theme={null}
    uv pip install -U "agno[os]" openai
    ```
  </Step>

  <Step title="Export your OpenAI API key">
    <CodeGroup>
      ```bash Mac/Linux theme={null}
      export OPENAI_API_KEY="your_openai_api_key_here"
      ```

      ```bash Windows theme={null}
      $Env:OPENAI_API_KEY="your_openai_api_key_here"
      ```
    </CodeGroup>
  </Step>

  <Step title="Run the example">
    Save the code above as `with_input_schema.py`, then run:

    ```bash theme={null}
    python with_input_schema.py
    ```
  </Step>
</Steps>

Full source: [cookbook/05\_agent\_os/09\_serving\_workflows/with\_input\_schema.py](https://github.com/agno-agi/agno/blob/v3.0.4/cookbook/05_agent_os/09_serving_workflows/with_input_schema.py)