Skip to main content
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

  1. Write the script and (optionally) attach datastores or accessories.
  2. Upload it — see Uploading a workflow. Maitai registers it under a workflow_ref_name.
  3. 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

​
model
string
required
The workflow to invoke, in the format workflow:<ref_name>.
​
messages
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."}
]
​
input
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"}
​
session_id
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.
​
stream
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.
​
secrets
object
Optional key-value pairs available to the workflow at runtime (e.g. API keys the workflow needs to call external services). Not stored.
​
metadata
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-..."
}
​
id
string
Unique identifier for the completion.
​
model
string
The workflow that was invoked.
​
choices
array
Array containing the workflow output. choices[0].message.content holds the result.
​
request_id
string
Maitai request identifier, useful for debugging.

​
Authentication

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

​
Error handling

StatusMeaning
400Invalid request (missing both input and messages, bad model format)
401Invalid or missing API key
404Workflow not found for your account
408Workflow exceeded its timeout
500Workflow execution error (check error.message for details)
Error responses follow the standard format: 
{
  "error": {
    "message": "Workflow not found: my-workflow",
    "type": "invalid_request_error"
  }
}

​
Next