Creating Workflows

Workflows are Python handlers that run through Centaur's durable workflow engine. They are useful when the task is longer than one agent turn: polling, branching, retries, waiting for external events, or coordinating multiple agent runs.

Use a workflow when the system needs durable progress rather than a single request-response turn. Common examples include scheduled reports, ETL syncs, incident monitors, approval gates, webhook-driven triage, long-running research jobs, and multi-agent handoffs that need to survive deploys or sandbox restarts.

Put organization workflows in an overlay repo under workflows/. See Using an overlay for packaging, mount paths, and chart configuration.

Migrating existing workflows to the api-rs Absurd runtime? See Workflows v2 Migration.

Workflows are loaded from WORKFLOW_DIRS. In an overlay deployment, workflow files must exist under the source's workflowsSubdir — by default workflows/ — in its repo-cache checkout, for example /var/lib/centaur/repos/your-org/centaur-overlay/workflows in the API container. Workflow-host sandboxes receive the same ordered list translated to /home/agent/github/.... Files in those directories are loaded the same way as built-in workflows; sources without the directory are skipped.

Define a workflow

Each workflow file exports WORKFLOW_NAME and an async handler(params, ctx). An optional Input dataclass gives structured inputs.

from dataclasses import dataclass
from datetime import timedelta
from typing import Any
 
from api.workflow_engine import WorkflowContext
 
 
WORKFLOW_NAME = "nightly_report"
 
WORKFLOW_PRINCIPAL = True
 
 
@dataclass
class Input:
    channel: str
    topic: str
 
 
async def handler(inp: Input, ctx: WorkflowContext) -> dict[str, Any]:
    data = await ctx.step("collect", lambda: {"topic": inp.topic})
    await ctx.sleep("settle", timedelta(seconds=30))
    result = await ctx.run_agent(
        "summarize",
        text=f"Write a short report about {data['topic']}",
    )
    return {"channel": inp.channel, "report": result}

WORKFLOW_PRINCIPAL is optional. Use it when the workflow host calls tools directly with ctx.call_tool(...) and should have its own credential boundary. Set it to True to have the API derive and register the workflow-nightly-report principal from WORKFLOW_NAME. Set it to an existing principal foreign ID, such as WORKFLOW_PRINCIPAL = "finance-automation", or to a prn_-prefixed OID to run the workflow-host sandbox under that principal instead. An unknown principal reference fails startup. Grant the required tool roles or secrets to the selected principal. Any WORKFLOW_PRINCIPAL value requires apiRs.workflowHostSandbox=true, which renders WORKFLOW_HOST_SANDBOX=true; startup fails if workflow-host sandboxing is disabled.

Durable primitives

The handler may re-execute after a restart. Put external side effects behind ctx.step(...) so completed work is not repeated.

These primitives compose into larger automations:

• Scheduled operations: run a daily digest, weekly cleanup, periodic sync, or business-hours monitor without a human prompt.
• Polling loops: sleep between checks for CI, blockchain confirmations, billing state, deploy health, or vendor exports.
• Event-driven flows: wait for a webhook, approval, upload, or callback and continue from the last checkpoint.
• Fan-out/fan-in orchestration: run independent named agents concurrently, then combine their successful results into one final result.
• Agent orchestration: use agents for judgment-heavy steps while the workflow owns timing, retries, state, and final delivery.

Run Agents Concurrently

Use ctx.run_agents(...) when independent reviewers or researchers should run at the same time:

reviews = await ctx.run_agents(
    [
        {"name": "correctness", "text": "Review the PR for correctness."},
        {"name": "security", "text": "Review the PR for security issues."},
        {"name": "tests", "text": "Review the PR's test coverage."},
    ],
    max_concurrency=3,
)

Every agent needs a unique, non-empty name. Centaur assigns each one a separate workflow-owned session and stable idempotency keys. The maximum concurrency defaults to 4 and may be set from 1 through 16. A batch supports up to 32 agents.

The result preserves input order and reports individual failures without discarding successful reviews:

{
  "results": [
    {"index": 0, "name": "correctness", "ok": true, "result": {"result_text": "..."}},
    {"index": 1, "name": "security", "ok": false, "error": "agent unavailable"},
    {"index": 2, "name": "tests", "ok": true, "result": {"result_text": "..."}}
  ],
  "succeeded": 2,
  "failed": 1
}

Batch items accept the same model, provider, reasoning, harness, persona, principal, prompt, content, idle timeout, maximum duration, and metadata options as ctx.agent_turn(...). Session and idempotency fields are reserved because the batch runtime assigns them independently for each agent.

Set principal to an existing principal foreign ID when an agent turn needs a specific credential boundary:

result = await ctx.agent_turn(
    "Prepare the finance report.",
    principal="finance-automation",
)

The principal is resolved before the session is created. An unknown or empty foreign ID fails the turn instead of using the thread-derived principal. If the session already exists under another principal, the turn returns a conflict instead of rebinding it.

Run a workflow

Direct administrative API calls need a short-lived Console service token. Mint one from the running Console instead of storing a separate static key:

CENTAUR_API_TOKEN=$(kubectl exec -n centaur-system deploy/centaur-centaur-console -- \
  bin/rails runner 'print ApiServer::Jwt.encode_for_console_service')

Create a run through the API:

curl -s "$CENTAUR_API_URL/api/workflows/runs" \
  -H "Authorization: Bearer $CENTAUR_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "workflow_name": "nightly_report",
    "input": {"channel": "ops", "topic": "open incidents"},
    "eager_start": true
  }' | jq

Inspect it:

curl -s "$CENTAUR_API_URL/api/workflows/runs/$RUN_ID" \
  -H "Authorization: Bearer $CENTAUR_API_TOKEN" | jq

List recent runs, optionally filtering by exact workflow name. The default limit is 50 and the maximum is 1000:

curl -s "$CENTAUR_API_URL/api/workflows/runs?workflow_name=nightly_report&limit=1000" \
  -H "Authorization: Bearer $CENTAUR_API_TOKEN" | jq

Start Workflows From Slack Buttons

Use ctx.slack_buttons to post a message whose buttons start a named workflow. The posting workflow receives the Slack message response and continues immediately.

from api.workflow_engine import Button
 
await ctx.slack_buttons(
    "release-review",
    channel="releases",
    text="Approve this release?",
    workflow="review_release",
    input={"release_id": "release-42"},
    buttons={
        "approve": Button("Approve", style="primary"),
        "reject": Button("Reject", style="danger"),
    },
)

primary uses Slack's green style, and danger uses its red style. Omit style for the default appearance. Plain string labels are still supported and can be mixed with Button values. Slack does not support arbitrary button colors.

Each click starts review_release with the supplied input and a click field. The Rust workflow context signs the target workflow, input, action, group, and destination channel before posting. The invoke endpoint verifies that signature before creating a run. Posting a lookalike button through the agent Slack tool does not grant permission to start workflows. Set channel to a channel name such as releases, or a Slack conversation ID such as C123 or D123. Names are resolved through the Slack tool's existing channel resolver before signing, so name lookup uses the same tool access as slack.send_message.

Slackbot supplies click from the verified Slack interaction, and the API uses it instead of any click field in the signed input. It contains id (the button group ID), action, user_id, team_id, channel_id, message_ts, and action_ts. A workflow using a dataclass Input can type this field as ButtonClick from api.workflow_engine.

The receiving workflow owns authorization, the message update, and the action:

WORKFLOW_NAME = "review_release"
 
async def handler(inp, ctx):
    click = inp["click"]
    if click["user_id"] != "U123" or click["action"] not in {"approve", "reject"}:
        return
 
    await ctx.step(
        "update-message",
        lambda: ctx.update_slack(
            click["channel_id"], click["message_ts"],
            text=f"Release {inp['release_id']}: {click['action']} selected.",
            blocks=[],
        ),
    )
    if click["action"] == "approve":
        await ctx.start_workflow(
            "deploy_release",
            {"release_id": inp["release_id"]},
            idempotency_key=f"deploy:{inp['release_id']}",
        )

Check team and channel as well when your authorization policy requires them. An unauthorized click can return without updating the message, leaving its buttons available to an authorized user. The receiving workflow must update or remove buttons explicitly. There is no automatic expiration or cancellation cleanup after posting.

Slack redelivery of the same click reuses the workflow execution through the engine's existing idempotency support. Distinct clicks start distinct executions, including clicks on different buttons in the same group. If only one operation should take effect, enforce that in the action, such as the stable child-workflow idempotency key above. External side effects inside steps must also be idempotent. Removing buttons does not cancel clicks that are already in flight.

Signatures do not prevent replay. Anyone who can read a signed button and post as the bot can copy it into another message in the same channel. Clicking that copy can start another execution with the same workflow and input. Signatures have no expiry and are not bound to the original message timestamp. The receiving workflow must check whether the action is still valid. For actions that should only happen once, use an idempotency key based on the business operation, such as deploy:release-42, rather than the click or receiving workflow execution.

Keep posting step names stable and unique. Posting uses an ordinary checkpoint and a stable Slack client_msg_id; an ambiguous Slack failure can still produce duplicate messages. All copies carry the same workflow target and group ID. Groups support one to five buttons. Action keys use letters, digits, underscores, or hyphens and contain at most 64 characters. Labels contain at most 75 characters. The signed button value, including its authentication envelope, must fit within 2000 bytes. The API rejects oversized values before posting. Pass record IDs rather than large or sensitive data because this input is stored in Slack.

This uses the existing Slackbot interactivity endpoint and API Slack bot token. Slackbot needs its dedicated SLACKBOT_API_KEY to start workflows through the button endpoint. The signing key stays in Rust and is derived specifically for workflow buttons from CENTAUR_JWT_SIGNING_SECRET. Unsigned buttons are rejected; rotating that secret invalidates previously signed buttons. Signatures authenticate the button configuration, while user authorization and duplicate-action policy remain the receiving workflow's responsibility. Workflow-start permissions still govern ordinary API callers. Existing custom Slack action IDs continue to emit raw workflow events. The centaur.workflow.action: namespace is reserved for workflow buttons. SLACK_API_URL can target a local Slack emulator in the API and Slackbot.

Schedule a workflow

Workflows can run from schedule metadata declared beside the handler. Use this when a workflow should be started by the platform on a clock instead of by an API call or webhook.

WORKFLOW_NAME = "daily_market_digest"
 
SCHEDULE = {
    "type": "cron",
    "cron": "0 9 * * MON-FRI",
    "timezone": "America/New_York",
    "input": {
        "channel": "markets",
        "topic": "overnight market structure and portfolio-relevant news",
    },
}

Cron schedules use five fields:

minute hour day-of-month month day-of-week

Examples:

Always set timezone for human-facing schedules. Without an explicit timezone, cron expressions are easy to misread across daylight saving changes and deployments in different regions.

Use input to keep the handler deterministic for scheduled runs: channel names, query scopes, tenant IDs, lookback windows, and delivery settings should be declared in the schedule instead of inferred from wall-clock state when possible.

For workflows that may run longer than their schedule interval, make each tick idempotent. Put writes and external API calls in named ctx.step(...) blocks, derive stable keys from the scheduled window, and have the handler detect already-processed periods before starting expensive work.

Interval schedules are useful when exact wall-clock alignment does not matter:

SCHEDULE = {
    "type": "interval",
    "seconds": 300,
    "input": {"target": "production"},
}

Use cron for calendar semantics such as "weekday at 9 AM"; use intervals for continuous monitors such as "check every five minutes".

Expose a workflow as a webhook

Workflows are private unless the workflow file explicitly exports WEBHOOKS. Each webhook is mounted at POST /api/webhooks/{slug} and creates a durable workflow run with a normalized webhook envelope. Use this for provider-driven entrypoints such as GitHub issue triage, billing events, or deploy callbacks.

from typing import Any
 
from api.webhooks import HeaderTriggerKey, HmacAuth, WebhookSpec
from api.workflow_engine import WorkflowContext
 
 
WORKFLOW_NAME = "github_issue_triage"
 
WEBHOOKS = [
    WebhookSpec(
        slug="github-issue-triage",
        provider="github",
        auth=HmacAuth.github(secret_ref="GITHUB_WEBHOOK_SECRET"),
        trigger_key=HeaderTriggerKey("X-GitHub-Delivery"),
        allowed_methods=["POST"],
        allowed_content_types=[
            "application/json",
            "application/x-www-form-urlencoded",
        ],
    )
]
 
 
async def handler(inp: dict[str, Any], ctx: WorkflowContext) -> dict[str, Any]:
    webhook = inp["webhook"]
    headers = webhook["headers"]
    payload = webhook["body"]
 
    if headers.get("x-github-event") != "issues":
        return {"skipped": True, "reason": "unsupported_event"}
 
    issue = payload["issue"]
    repo = payload["repository"]["full_name"]
    result = await ctx.agent_turn(
        f"Triage GitHub issue {repo}#{issue['number']}: {issue['title']}",
        thread_key=f"github:{repo}:{issue['number']}",
    )
    return {"triaged": True, "agent_result": result}

Configure the provider to call:

https://<your-centaur-host>/api/webhooks/github-issue-triage

For GitHub, set the webhook secret to the same value as GITHUB_WEBHOOK_SECRET in the API deployment and select application/json. GitHub's default application/x-www-form-urlencoded payloads also work when that content type is listed in allowed_content_types.

Webhook requests do not use Centaur API keys. The API verifies the provider signature before creating workflow state. HmacAuth.github(...) verifies X-Hub-Signature-256; a plain HmacAuth(...) can be used for other SHA-256 HMAC providers that sign the raw request body. During local development or for trusted internal routes, auth="none" is allowed.

For providers that implement the Standard Webhooks specification, use the explicit standard_webhooks auth type:

WEBHOOKS = [
    {
        "slug": "feed-ingest",
        "provider": "feed-provider",
        "auth": {
            "type": "standard_webhooks",
            "secret_ref": "FEED_WEBHOOK_SECRET",
        },
        "trigger_key": {"type": "header", "header": "webhook-id"},
        "allowed_methods": ["POST"],
        "allowed_content_types": ["application/json"],
    }
]

Set FEED_WEBHOOK_SECRET in the API deployment to the provider's signing secret (normally a whsec_-prefixed, base64-encoded value). Centaur requires the webhook-id, webhook-timestamp, and webhook-signature headers, verifies the v1 signature over the exact raw body, supports space-separated signatures for key rotation, and rejects timestamps outside a five-minute window. Using webhook-id as the trigger key also gives retries the same durable workflow identity. Auth schemes are selected by auth.type; Centaur does not infer one from the headers.

The workflow receives input in this shape:

{
  "webhook": {
    "slug": "github-issue-triage",
    "provider": "github",
    "method": "POST",
    "path": "/api/webhooks/github-issue-triage",
    "headers": {
      "x-github-event": "issues",
      "x-github-delivery": "..."
    },
    "query": {},
    "body": {},
    "raw_body_sha256": "...",
    "source_ip": "203.0.113.10"
  }
}

Sensitive headers such as signatures, cookies, authorization, and API keys are removed before the workflow input is persisted. trigger_key controls idempotency; prefer a provider delivery header like X-GitHub-Delivery. If no trigger key is configured, Centaur uses the raw body SHA-256 hash.

The webhook endpoint returns 202 when it creates a new run and 200 when the same trigger key maps to an existing run.

Verify

After deploying an overlay, check API logs for workflow load events and create a small run with eager_start: true. If the workflow is missing, inspect WORKFLOW_DIRS, the configured repo/ref in repo-cache, and whether the file exports WORKFLOW_NAME. For webhooks, also check for workflow_webhook_registered in the API logs and send a signed request to the public /api/webhooks/{slug} URL.