Workflows - Maitai

Maitai Workflows let you execute multi-step AI pipelines through a single API call. A workflow is a Python script you write; Maitai runs it in a managed worker pool and handles execution, scaling, monitoring, and caching. Orchestrate models, agents, HTTP services, and custom logic, then invoke the whole thing the way you’d call any model.

What a workflow is

A workflow is one Python file with an execute(ctx) function. Maitai calls execute, passes it a context object, and returns whatever you return.

from maitai_workflow import WorkflowContext

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

The ctx object is the workflow’s connection to everything: its input, Maitai model and agent calls, arbitrary HTTP, queryable datastores, streaming, and logging.

Building blocks

Structure

The script layout, the execute entrypoint, input/output schemas, and helpers.

Context (ctx)

The runtime API, models, agents, HTTP, streaming, logging, and more.

Datastores & Accessories

Attach reference data, bundled files or large, queryable record stores.

Invoke

Call a workflow from the SDK or raw API, including streaming.

Lifecycle

Write the script and (optionally) attach datastores or accessories.
Upload it, see Uploading a workflow. Maitai registers it under a workflow_ref_name.
Invoke it as model="workflow:<ref_name>" (below). Maitai runs it, monitors every model call, and returns the result.

The Maitai team can also configure workflows for your account during onboarding.

Invoke a workflow

Set model to workflow:<your-workflow-name>. You can pass either messages (chat-style) or input (arbitrary JSON payload), or both.

With messages (chat-style)

from maitai import Maitai

maitai = Maitai()

response = maitai.workflow.completions.create(
    model="workflow:my-workflow",
    messages=[
        {"role": "user", "content": "Summarize this quarter's earnings report."},
    ],
)
print(response.choices[0].message.content)

import Maitai from "maitai";

const maitai = new Maitai();

const response = await maitai.workflow.completions.create({
  model: "workflow:my-workflow",
  messages: [
    { role: "user", content: "Summarize this quarter's earnings report." },
  ],
});

console.log(response.choices[0].message.content);

With input (structured payload)

For workflows that expect structured data rather than chat messages, use input:

response = maitai.workflow.completions.create(
    model="workflow:my-etl-pipe",
    input={"document_url": "https://...", "output_format": "markdown"},
)

const response = await maitai.workflow.completions.create({
  model: "workflow:my-etl-pipe",
  input: { document_url: "https://...", output_format: "markdown" },
});

Raw API (cURL)

curl -X POST https://api.trymaitai.ai/workflow/chat/completions \
  -H "Content-Type: application/json" \
  -H "x-api-key: $MAITAI_API_KEY" \
  -d '{
    "model": "workflow:my-workflow",
    "input": {"document_url": "https://...", "output_format": "markdown"}
  }'

With session_id (for correlation)

Pass session_id to correlate all model calls within a workflow run in Maitai analytics:

response = maitai.workflow.completions.create(
    model="workflow:my-workflow",
    messages=[{"role": "user", "content": "..."}],
    session_id="my-session-123",  # Optional; overrides auto-generated
)

Streaming responses

To receive workflow output in real-time as the script executes, set stream=True. The workflow will yield intermediate chunks as they are emitted (via ctx.emit()), followed by the final completion.

stream = maitai.workflow.completions.create(
    model="workflow:my-workflow",
    input={"document_url": "https://...", "output_format": "markdown"},
    stream=True,
)

for chunk in stream:
    if hasattr(chunk, "type"):
        # This is an intermediate workflow chunk (from ctx.emit)
        print(f"[{chunk.type}] {chunk.content}")
    elif hasattr(chunk, "choices"):
        # This is the final ChatCompletion object
        print("Final result:", chunk.choices[0].message.content)

const stream = await maitai.workflow.completions.create({
  model: "workflow:my-workflow",
  input: { document_url: "https://...", output_format: "markdown" },
  stream: true,
});

for await (const chunk of stream) {
  if (chunk.type) {
    // Intermediate chunk
    console.log(`[${chunk.type}]`, chunk.content);
  } else if (chunk.choices) {
    // Final completion
    console.log("Final result:", chunk.choices[0].message.content);
  }
}

Passing secrets

If your workflow requires runtime credentials (e.g. third-party API keys), pass them via secrets. These are never stored, they’re only available for the duration of the request.

response = maitai.workflow.completions.create(
    model="workflow:my-workflow",
    messages=[{"role": "user", "content": "Fetch and summarize the latest data."}],
    secrets={"EXTERNAL_API_KEY": "tok-..."},
)

const response = await maitai.workflow.completions.create({
  model: "workflow:my-workflow",
  messages: [{ role: "user", content: "Fetch and summarize the latest data." }],
  secrets: { EXTERNAL_API_KEY: "tok-..." },
});

Request format

string

required

The workflow to invoke, in the format workflow:<ref_name>.

array

An array of chat messages (OpenAI format). Required unless input is provided.

[
  {"role": "system", "content": "You are an analyst."},
  {"role": "user", "content": "Summarize the report."}
]

object | string | array

Arbitrary JSON payload passed to the workflow. Required unless messages is provided. Use this for Lambda-style invocations where chat messages don’t apply.

{"document_url": "https://...", "output_format": "markdown"}

string

Optional session ID for the workflow run. If provided, overrides the auto-generated session ID. All model calls within the workflow use this session, enabling correlation across steps in Maitai analytics.

boolean

Optional. When true, returns a stream of intermediate workflow chunks (emitted via ctx.emit()) followed by a final chat.completion object, allowing you to display real-time progress. Default false.

object

Optional key-value pairs available to the workflow at runtime (e.g. API keys the workflow needs to call external services). Not stored.

object

Optional metadata passed to the workflow.

Response format

Workflows return a standard OpenAI-compatible chat.completion response:

{
  "id": "chatcmpl-abc123",
  "object": "chat.completion",
  "created": 1770848541,
  "model": "workflow:my-workflow",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "Here is the earnings summary..."
      },
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 0,
    "completion_tokens": 0,
    "total_tokens": 0
  },
  "request_id": "d4f7e2a1-..."
}

string

Unique identifier for the completion.

string

The workflow that was invoked.

array

Array containing the workflow output. choices[0].message.content holds the result.

string

Maitai request identifier, useful for debugging.

Authentication

Workflows use the same API key authentication as all Maitai endpoints. See Authentication.

Error handling

Status	Meaning
`400`	Invalid request (missing both `input` and `messages`, bad `model` format)
`401`	Invalid or missing API key
`404`	Workflow not found for your account
`408`	Workflow exceeded its timeout
`500`	Workflow execution error (check `error.message` for details)

Error responses follow the standard format:

{
  "error": {
    "message": "Workflow not found: my-workflow",
    "type": "invalid_request_error"
  }
}

Write your first workflow: Workflow Structure
The full runtime API: Workflow Context (ctx)
Attach reference data and upload: Datastores & Accessories

​What a workflow is

​Building blocks

Structure

Context (ctx)

Datastores & Accessories

Invoke

​Lifecycle

​Invoke a workflow

​With messages (chat-style)

​With input (structured payload)

​Raw API (cURL)

​With session_id (for correlation)

​Streaming responses

​Passing secrets

​Request format

​Response format

​Authentication

​Error handling

​Next

What a workflow is

Building blocks

Lifecycle

Invoke a workflow

With messages (chat-style)

With input (structured payload)

Raw API (cURL)

With session_id (for correlation)

Streaming responses

Passing secrets

Request format

Response format

Authentication

Error handling

Next