Tools and external steps

Some steps in the middle of a pipeline aren't decisions, they're lookups: a search, a normalisation, an insert. Mark one type: tool and Otium calls your endpoint with the step's input, then takes the JSON it returns as the step's output, which flows into the next step like a model's would. No cron job on your side, and nothing waiting for one.

name: match-product
description: Classify a listing, search your catalog, pick the match, record it.

# The endpoint every tool node in this workflow calls, registered once by name.
tools: { endpoint: catalog-api }

input:
  schema:
    type: object
    required: [title]
    properties:
      title: { type: string }

schemas:
  category:
    type: object
    required: [slug]
    properties:
      slug: { type: string }
  candidates:
    type: object
    required: [skus]
    properties:
      skus: { type: array, items: { type: string } }
  match:
    type: object
    required: [sku]
    properties:
      sku: { type: string }

nodes:
  classify:
    target: { product: otium-small }
    estimate: { tokens_in: 200, tokens_out: 20 }
    body:
      messages:
        - role: user
          content: "Reply as JSON with one category slug for: ${input.title}"
    output:
      validators:
        - { type: json_schema, schema: { $ref: "#/schemas/category" } }
      on_invalid: { retry: 1 }

  search:
    type: tool
    tool: { op: candidate_search, timeout: 1s, idempotent: true }
    input: # the request body your endpoint receives, as JSON
      slug: "${nodes.classify.output.slug}"
      title: "${input.title}"
    output:
      validators:
        - { type: json_schema, schema: { $ref: "#/schemas/candidates" } }
      on_invalid: { retry: 1 }

  choose:
    target: { product: otium-small }
    estimate: { tokens_in: 400, tokens_out: 20 }
    body:
      messages:
        - role: user
          content: "Which of ${nodes.search.output.skus} is ${input.title}? Reply as JSON with the sku."
    output:
      validators:
        - { type: json_schema, schema: { $ref: "#/schemas/match" } }
      on_invalid: { retry: 1 }

  record:
    type: tool
    tool: { op: record_match, timeout: 2s } # a write: not idempotent, so called once
    input:
      sku: "${nodes.choose.output.sku}"
      title: "${input.title}"

search reads your catalog, so it is marked idempotent and may be retried. record writes, so it is not, and Otium calls it once. op is what your handler switches on; it defaults to the step's name.

A workflow names one endpoint in tools: { endpoint }, and every tool step in it calls that endpoint with its own op, so one handler serves the whole workflow. Register it once by name:

otiumctl tools register --name catalog-api --url https://tools.example.com/otium
# or: POST /v1/tools  {"name": "catalog-api", "url": "https://tools.example.com/otium"}

Registering prints the endpoint's signing secret once; keep it. Publishing a workflow that names an endpoint you haven't registered is refused. The endpoint must be HTTPS on a public address, and redirects are not followed. GET /v1/tools lists your endpoints and the addresses our calls come from, so you can allow them through a firewall. otiumctl tools rotate issues a new secret.

One signed POST per call:

POST https://tools.example.com/otium
Content-Type: application/json
Otium-Signature: t=1760300000,v1=5f8a…
Otium-Tool-Op: candidate_search
Otium-Tool-Call: call_…

{"op":"candidate_search","call_id":"call_…","run_id":"run_…","node":"search",
 "input":{"slug":"widgets","title":"Blue widget, 3-pack"}}

Check the signature before you trust the body. It is an HMAC-SHA256 over the timestamp and the raw body, keyed with your endpoint's secret, the same scheme as our webhooks:

# Otium-Signature: t=<unix seconds>,v1=<hex>
expected = hex(HMAC_SHA256(secret, t + "." + raw_body))
accept if  v1 == expected  (constant-time compare)
       and |now - t| <= 5 minutes

Answer with a 2xx and a JSON body of up to 1 MB. That body becomes the step's output and is checked by its validators, exactly like a model's answer. A non-JSON body, a larger one, or any other status fails the call.

$0.002 per call your endpoint answers with a 2xx. A call that fails (a timeout, an error status, an address no call may reach) is not charged. An answer your validators then reject was still answered, so it is charged, and so is each idempotent retry that gets an answer. There are no tokens in a tool call, so the price is the same whatever the step's size or your window.

A call is made once unless you set idempotent: true. Not retried when its output is rejected, not retried after an infrastructure failure, never sent to a fallback: a step that writes can never write twice. If the call may already have reached you and we lose track of it, the step fails rather than risk a second call.

An idempotent call follows the step's retry policy, and it is also retried when your endpoint answers 5xx or 429, is unreachable, or times out. Any other status is taken as your endpoint saying no, and is not retried. call_id stays the same when one attempt is resent, so you can dedupe on it; a new attempt after a rejected output gets a new one.

The timeout defaults to 10 seconds and can be set up to 60. Work that takes longer belongs in an external step. A tool step's timeout is also what the compiler uses as its worst-case duration, so a tight timeout keeps the plan's estimate honest.

A failed call fails the step with its own reason, so you can tell your endpoint failing from a model failing: tool_timeout (no answer in time), tool_error (an error status, an unreadable answer, or a call we could not safely repeat) and tool_refused (an address no call may reach).

In unit tests a tool step takes canned answers like any other step:

cases:
  - name: a match is found and recorded
    input: { title: "Blue widget, 3-pack" }
    nodes:
      classify: { output: { slug: widgets } }
      search: { output: { skus: [W-100, W-200] } }
      choose: { output: { sku: W-200 } }
      record: { output: { recorded: true } }
    expect:
      status: succeeded
      succeeded: [classify, search, choose, record]
      prompts:
        choose: { contains: ["W-100", "W-200", "Blue widget, 3-pack"] }

  - name: a search answer that breaks the schema is retried once
    input: { title: "Blue widget, 3-pack" }
    nodes:
      classify: { output: { slug: widgets } }
      search:
        - output: { results: [] } # no `skus`: rejected by the validator
        - output: { skus: [W-200] }
      choose: { output: { sku: W-200 } }
      record: { output: { recorded: true } }
    expect:
      status: succeeded
      attempts: { search: 2, record: 1 }

  - name: a write that fails is not called again
    input: { title: "Blue widget, 3-pack" }
    nodes:
      classify: { output: { slug: widgets } }
      search: { output: { skus: [W-200] } }
      choose: { output: { sku: W-200 } }
      record: { fail: true }
    expect:
      status: partial # every model step succeeded; the write did not
      attempts: { record: 1 }
      failed: { record: tool_error }

Otium runs no code of yours. When your pipeline needs your own system, or a person, in the middle, mark that step type: external: the run parks, you do the work, and you resume it with the step's output. The run's execution clock stops while it waits, no machine is held for it, and it is still one run, so a pipeline that interleaves your steps with ours keeps a single completion window rather than one per segment.

name: reviewed-description
description: Draft a description, wait for your review, then write the final copy.

input:
  schema:
    type: object
    required: [listing]
    properties:
      listing: { type: string }

schemas:
  draft:
    type: object
    required: [text]
    properties:
      text: { type: string }
  verdict:
    type: object
    required: [approved, notes]
    properties:
      approved: { type: boolean }
      notes: { type: string }

nodes:
  draft:
    target: { product: otium-small }
    estimate: { tokens_in: 600, tokens_out: 200 }
    body:
      messages:
        - role: user
          content: "Draft a product description as JSON with a text field: ${input.listing}"
    output:
      validators:
        - { type: json_schema, schema: { $ref: "#/schemas/draft" } }
      on_invalid: { retry: 2 }

  review:
    type: external
    input: # what your side receives when the run parks here
      listing: "${input.listing}"
      draft: "${nodes.draft.output.text}"
    max_park: 72h # give up on the run if nobody answers by then
    park_estimate: { p50: 8h, p95: 24h } # how long your side usually takes
    output: # what you resume it with, checked like a model's output
      validators:
        - { type: json_schema, schema: { $ref: "#/schemas/verdict" } }

  final:
    target: { product: otium-medium }
    estimate: { tokens_in: 600, tokens_out: 250 }
    body:
      messages:
        - role: user
          content: "Rewrite this description. Reviewer notes: ${nodes.review.output.notes}. Draft: ${nodes.draft.output.text}"

input is what you receive when the run parks, rendered from the item and earlier steps. It is also how the step's dependencies are worked out, since it has no prompt to read them from. The output validators check what you resume it with, so malformed data from your side fails at the boundary instead of three steps later.

Give each external step two durations:

• max_park is the failsafe. If nobody resumes the step by then, it fails with park_expired, the steps that needed it fail with upstream_failed, and the run ends partial: everything that already succeeded is still delivered.
• park_estimate: { p50, p95 } is how long your side usually takes. Without it, a park is counted at its full max_park wherever a duration is estimated.

A parked run uses none of its execution budget, but the batch's wall-clock expiry keeps running. So when you submit, the workflow's p95 path, parks included, must fit before the batch expires: a review that takes up to a day does not fit a 24-hour window, and the batch is refused with expiry_too_short. Submit it on the 7-day window instead. See windows and the two clocks.

A batch of 10,000 items parks 10,000 runs at the same step, so parking is handled per batch, not per run. When runs first start waiting, Otium sends one run_set.node_parked notification. List the waiting runs, with each one's rendered input, 200 at a time:

GET /v1/run-sets/{id}/nodes/review/parked?limit=200&after={next_cursor}

{"run_set_id":"rs_…","node_key":"review","count":2,"next_cursor":"",
 "runs":[{"run_id":"run_…","custom_id":"sku-1001","parked_at":"…",
          "input":{"listing":"Blue widget, 3-pack","draft":"A blue widget."}}, …]}

The list is live: runs arrive at a park at different times, so you can resume in waves on your own schedule and nothing waits for the cohort to be complete. Resume them with one JSONL body, one line per run, the same shape as a batch file:

POST /v1/run-sets/{id}/nodes/review/resume

{"custom_id":"sku-1001","output":{"approved":true,"notes":"Mention the 3-pack."}}
{"custom_id":"sku-1002","output":{"approved":false,"notes":"Wrong colour in the photo."}}

{"resumed":1,"rejected":1,
 "results":[{"custom_id":"sku-1001","run_id":"run_…","ok":true},
            {"custom_id":"sku-1002","run_id":"run_…","ok":false,
             "error":"output rejected by the node's validators","verdict":{…}}]}

Each line is checked on its own. A line whose output the validators reject, or whose custom_id isn't in the batch, fails that line and never the whole request, and the rejected run stays parked so you can send it again. Re-sending the whole file after a timeout is safe: a run that already resumed is not resumed twice.

For one run, while debugging, resume it directly:

POST /v1/runs/{run_id}/nodes/review/resume
{"output":{"approved":true,"notes":"Mention the 3-pack."}}

GET /v1/runs/{run_id} shows a parked step's rendered input too, so a missed notification never means replaying the run. In unit tests the canned output of an external step is what it is resumed with, and lapse: true leaves it unanswered:

tier: remnant
cases:
  - name: the review comes back and the final copy uses its notes
    input: { listing: "Blue widget, 3-pack" }
    nodes:
      draft: { output: { text: "A blue widget." } }
      review: { output: { approved: true, notes: "Mention the 3-pack." } }
      final: { output: "Three blue widgets in one pack." }
    expect:
      status: succeeded
      prompts:
        final: { contains: ["Mention the 3-pack.", "A blue widget."] }

  - name: nobody answers the review
    input: { listing: "Blue widget, 3-pack" }
    nodes:
      draft: { output: { text: "A blue widget." } }
      review: { lapse: true }
    expect:
      status: partial # the draft is still yours
      succeeded: [draft]
      failed: { review: park_expired, final: upstream_failed }

Instead of polling a batch for hours or days, register an HTTPS endpoint and Otium tells you when something happens:

otiumctl webhooks add --url https://hooks.example.com/otium
# subscribes to run_set.completed and run_set.node_parked; add per-run events with --events
# or: POST /v1/webhooks  {"url": "https://hooks.example.com/otium", "events": [...]}

A new endpoint gets the two batch-level events. The per-run events fire once per item, so a 10,000-item batch sends 10,000 of each; subscribe to them only where that is what you want.

POST https://hooks.example.com/otium
Otium-Signature: t=1760300000,v1=…
Otium-Event: run_set.node_parked
Otium-Delivery: dlv_…
Otium-Attempt: 1

{"id":"evt_…","event":"run_set.node_parked","created_at":"…",
 "run_set_id":"rs_…","batch_id":"batch_…","workflow":"reviewed-description",
 "node_key":"review","parked":2,
 "url":"https://api.getotium.ai/v1/run-sets/rs_…/nodes/review/parked"}

A notification carries identifiers, counts and a typed status, never a prompt or a result: fetch those from the API it links to, with your API key. Verify Otium-Signature exactly as for tool calls. Otium-Delivery stays the same across retries, so you can dedupe on it.

Answer with a 2xx within 10 seconds. Anything else is retried after 30 seconds, 2 minutes, 10 minutes, 1 hour and 6 hours, then dropped. A failed notification never changes a run. otiumctl webhooks deliveries shows what we sent and what your endpoint answered, and disabling an endpoint pauses it without building a backlog.