Skip to content

External Workers#

External workers run handler code outside the engine. They poll the versioned REST API, execute a task, and report completion or failure. Use them to share application code and secrets, scale handler capacity independently, or implement handlers in any language that can send JSON.

Note
The protocol is at least once. A lease can expire after a worker performed an external action but before completion committed. Use a provider idempotency key and reconcile unknown outcomes; do not assume a handler runs exactly once.

Protocol#

Routes below are relative to /api/v1. Secure deployments require x-api-key andx-tenant-id on each request.

1. Poll#

POST/workers/tasks/poll
JSON
{
  "handler_name": "send_email",
  "worker_id": "mail-worker-42",
  "limit": 10,
  "version": "0.7.0"
}
ParameterTypeRequiredDescription
handler_namestringYesHandler this worker accepts.
worker_idstringYesUnique identity for this process.
limitintegerNoMaximum tasks to claim in this poll.
versionstringNoWorker version checked against fleet pins.

The response is an array. A reclaimed task may includeresume_checkpoint and its monotoniccheckpoint_seq. Resume from that checkpoint instead of repeating completed activity work.

2. Heartbeat and checkpoint#

POST/workers/tasks/{id}/heartbeat
JSON
{
  "worker_id": "mail-worker-42",
  "checkpoint_seq": 0,
  "checkpoint": {
    "completed_batches": 12,
    "cursor": "next-page-token"
  }
}

Send a heartbeat every 15–30 seconds for long tasks. The response contains the next checkpoint_seq. A stale sequence or former lease owner receives 409 Conflict. Checkpoints are capped at 256 KiB, encrypted at rest when storage encryption is enabled, and survive retry and stale-lease recovery.

3. Complete#

POST/workers/tasks/{id}/complete
JSON
{
  "worker_id": "mail-worker-42",
  "output": {
    "message_id": "msg-123",
    "delivered": true
  }
}

4. Fail#

POST/workers/tasks/{id}/fail
JSON
{
  "worker_id": "mail-worker-42",
  "message": "SMTP connection refused",
  "retryable": true
}

A retryable failure reschedules the instance according to the block retry policy. A permanent failure enters the surrounding TryCatch branch or, without one, the dead letter queue.

Runnable Python worker#

Python
import os
import socket
import time
import httpx

ENGINE = os.getenv("ORCH8_URL", "http://localhost:8080/api/v1")
API_KEY = os.environ["ORCH8_API_KEY"]
TENANT_ID = os.getenv("ORCH8_TENANT_ID", "demo")
WORKER_ID = f"py-worker-{socket.gethostname()}-{os.getpid()}"

def send_email(task):
    # Pass task["id"] or another stable value as the provider idempotency key.
    return {"message_id": "msg-123"}

with httpx.Client(
    base_url=ENGINE,
    timeout=30,
    headers={"x-api-key": API_KEY, "x-tenant-id": TENANT_ID},
) as http:
    while True:
        response = http.post("/workers/tasks/poll", json={
            "handler_name": "send_email",
            "worker_id": WORKER_ID,
            "limit": 5,
            "version": "0.7.0",
        })
        response.raise_for_status()

        for task in response.json():
            try:
                output = send_email(task)
                result = http.post(
                    f"/workers/tasks/{task['id']}/complete",
                    json={"worker_id": WORKER_ID, "output": output},
                )
                result.raise_for_status()
            except Exception as error:
                failure = http.post(
                    f"/workers/tasks/{task['id']}/fail",
                    json={
                        "worker_id": WORKER_ID,
                        "message": str(error),
                        "retryable": True,
                    },
                )
                failure.raise_for_status()

        time.sleep(1)

Resumable activity loop#

For batch work, restore the checkpoint from the claimed task and update it atomically as each durable unit completes:

Python
checkpoint = task.get("resume_checkpoint") or {
    "completed_batches": 0,
    "cursor": None,
}
checkpoint_seq = task.get("checkpoint_seq", 0)

while has_more(checkpoint["cursor"]):
    checkpoint = process_next_batch(checkpoint)
    heartbeat = http.post(
        f"/workers/tasks/{task['id']}/heartbeat",
        json={
            "worker_id": WORKER_ID,
            "checkpoint_seq": checkpoint_seq,
            "checkpoint": checkpoint,
        },
    )
    heartbeat.raise_for_status()
    checkpoint_seq = heartbeat.json()["checkpoint_seq"]
Warning
A 409 heartbeat means this process no longer owns the current lease or used a stale checkpoint sequence. Stop work and do not report completion.

Fleet controls#

  • GET /workers andGET /workers/tasks/stats expose registrations and queue state.
  • POST /workers/commands sends drain, reload, or ping commands; workers acknowledge commands after applying them.
  • POST /workers/version-pins sets a minimum version for a tenant and handler.
  • Named queues use POST /workers/tasks/poll/queue. Push queues deliver a signed envelope to the configured worker URL.

Use @orch8.io/sdk for the official Node client, or implement the REST protocol directly as above.

Ready to try Orch8?

One command to install. Then run your first local sequence.

Bash
curl -fsSL https://orch8.io/start.sh | sh
Quick Start GuideOr skip the setup with Cloud →
Previous← Template InterpolationNextConcurrency & Rate Limits →