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

# Workflow Structure

> How a workflow script is organized — the execute entrypoint, input/output schemas, and helpers.

A Maitai workflow is a single Python file. Maitai loads it into the [workflow worker pool](/build/workflows/overview), calls one function — `execute(ctx)` — and returns whatever that function returns. Everything you need at runtime arrives on the [`ctx` object](/build/workflows/context): input, secrets, LLM access, HTTP, datastores, and more.

Here is a complete, minimal workflow:

```python theme={null}
from maitai_workflow import WorkflowContext

APPLICATION = "My Application"

INPUT_SCHEMA = {
    "query": {"type": "string", "description": "The user's question", "required": True},
    "context": {"type": "string", "description": "Optional background context"},
}

OUTPUT_SCHEMA = {
    "answer": {"type": "string", "description": "The generated answer"},
    "confidence": {"type": "number", "description": "Confidence between 0 and 1"},
}


def execute(ctx: WorkflowContext):
    inp = ctx.input or {}
    query = inp.get("query", "")
    if not query:
        return {"error": "query is required"}

    resp = ctx.chat.completions.create(
        messages=[{"role": "user", "content": query}],
        application=APPLICATION,
        intent="answer_query",
    )
    answer = resp["choices"][0]["message"]["content"]

    return {"answer": answer, "confidence": 0.85}
```

<Note>
  The repo ships this as a copy-paste starting point at `workflows/workflow.example.py`. You can also fetch the latest template from the API at `GET /workflows/default-script`.
</Note>

## The `execute(ctx)` entrypoint

Every workflow **must** define exactly one top-level `execute` function.

* **Required and unique.** There must be precisely one module-level `def execute(...)`. Upload validation rejects a script with zero or multiple.
* **Synchronous.** Write `def execute(ctx):`, not `async def`. The worker runs your function on a thread — use `ctx`'s synchronous methods (they handle their own I/O) rather than `await`.
* **Takes the context.** It receives a single argument, the [`WorkflowContext`](/build/workflows/context), conventionally named `ctx`.
* **Returns a value.** Return your result — typically a `dict` matching `OUTPUT_SCHEMA`. A bare `return` or `return None` is rejected at upload. The returned object is exposed verbatim as `response.output`, and a JSON-stringified copy becomes `choices[0].message.content` for OpenAI-shaped consumers.

You can define as many helper functions, constants, and imports at module level as you like — only `execute` is special. Keep helpers as plain top-level `def`s; the [flow graph](/build/workflows/overview) inlines them when visualizing your workflow.

```python theme={null}
SYSTEM_PROMPT = "You are a careful financial analyst."

def _build_messages(query: str) -> list:
    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": query},
    ]

def execute(ctx: WorkflowContext):
    resp = ctx.chat.completions.create(
        messages=_build_messages(ctx.input["query"]),
        application="Finance", intent="analyze",
    )
    return {"answer": resp["choices"][0]["message"]["content"]}
```

## The `WorkflowContext` import

```python theme={null}
from maitai_workflow import WorkflowContext
```

If you reference `WorkflowContext` anywhere — most commonly as the type hint on `execute(ctx: WorkflowContext)` — you must import it, or the script fails at runtime with a `NameError`. Upload validation flags a missing import. The import is also what makes editor autocomplete work for the full `ctx` surface.

## Input and output schemas

`INPUT_SCHEMA` and `OUTPUT_SCHEMA` are optional module-level dicts that describe the shape of `ctx.input` and your return value.

```python theme={null}
INPUT_SCHEMA = {
    "document_url": {"type": "string", "description": "URL of the PDF to process", "required": True},
    "output_format": {"type": "string", "description": "markdown | html"},
    "max_pages": {"type": "number", "description": "Cap on pages to read"},
}
```

* **Format.** A dict mapping each field name to `{type, description, required?}`.
* **Types.** `"string"`, `"number"`, `"boolean"`, `"object"`, `"array"`.
* **Required fields.** Add `"required": True`. Omitting the key (or `False`) marks the field optional.

<Note>
  Schemas are **informational, not enforced at runtime.** They power the Portal — describing what `ctx.input` expects and what your output contains, and rendering the **Input** and **Output** nodes in the flow diagram. Validate inputs you actually depend on inside `execute`.
</Note>

The upload step extracts these dicts statically (via AST), so they must be **literal** dicts at module level — don't compute them at import time or build them inside a function.

## Recognized constants

| Constant        | Purpose                                                                                                                                                                                                                                                                                                                   |
| --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `APPLICATION`   | The application this workflow's LLM calls belong to. Used as the default `application` for `ctx.chat.completions.create(...)` when you pass it explicitly, and it documents intent. Binding is set at upload via `--application-name` / `--application-id` (see [Uploading](/build/workflows/data#uploading-a-workflow)). |
| `INPUT_SCHEMA`  | Input shape (above).                                                                                                                                                                                                                                                                                                      |
| `OUTPUT_SCHEMA` | Output shape (above).                                                                                                                                                                                                                                                                                                     |

Any other module-level constants (like `SYSTEM_PROMPT` above) are yours to use freely — they aren't special to Maitai.

## Streaming output

If callers invoke your workflow with `stream=True`, you can push intermediate events to them as the script runs by calling [`ctx.emit(...)`](/build/workflows/context#ctx-emit). When not streaming, emitted chunks are buffered and returned with the final response, so the same code works either way.

```python theme={null}
def execute(ctx: WorkflowContext):
    ctx.emit({"type": "status", "content": "fetching document"})
    doc = ctx.get(ctx.input["document_url"])

    ctx.emit({"type": "status", "content": "summarizing"})
    resp = ctx.chat.completions.create(
        messages=[{"role": "user", "content": doc["text"]}],
        application="Docs", intent="summarize",
    )
    return {"summary": resp["choices"][0]["message"]["content"]}
```

See [Invoking workflows — streaming responses](/build/workflows/overview#streaming-responses) for the consumer side.

## `transform_session` (optional)

Define `transform_session(session)` if you want to convert recorded Maitai sessions into [test set](/test/test_sets/creation) items for this workflow. It takes a session and returns the `input`/`output` pair that `execute` would have seen:

```python theme={null}
def transform_session(session):
    workflow_input, workflow_output = {}, {"content": ""}
    if session.timeline:
        first = session.timeline[0]
        if first.request and first.request.params:
            for msg in reversed(first.request.params.messages):
                if msg.role == "user" and msg.content:
                    workflow_input = {"query": msg.content}
                    break
        last = session.timeline[-1]
        if last.response and last.response.choices:
            workflow_output = {"content": last.response.choices[0].message.content or ""}
    return {"input": workflow_input, "output": workflow_output}
```

The return value must be a dict with exactly two keys, `"input"` and `"output"` (where `"output"` is itself a dict). When present, the Portal lets you import sessions into a test set; when absent, that option is simply hidden.

## Next

* The full runtime API: [Workflow Context (`ctx`)](/build/workflows/context)
* Attach reference data: [Datastores & Accessories](/build/workflows/data)
* Call your workflow from code: [Invoking workflows](/build/workflows/overview)