Guides

Scale out with worker nodes

Run agent execution on multiple machines: start HTTP worker nodes on port 7999, health-check them from a coordinator, and let DistributedExecutor dispatch every task to the least-loaded healthy node — with worker-level circuit breakers handling the failover.

Route map

1. 1

   Start a worker

   Launch the HTTP worker server with replicate-agent workers start or serve_worker().

2. 2

   Health-check from the coordinator

   Confirm reachability with curl /health and replicate-agent workers ping before writing any coordinator code.

3. 3

   Build the coordinator

   Wire RemoteWorkerNode + HttpWorkerTransport into a DistributedExecutor, with a local fallback node.

4. 4

   Verify failover

   Understand how worker circuit breakers report state via /health and how routing skips OPEN nodes.

Coordinator and workers: the topology

The distributed layer has exactly two roles. A worker executes agent invocations — it owns an AgentExecutor and talks to the Replicate API. A coordinator owns a DistributedExecutor and decides which worker gets each task. Workers come in two flavors:

• Local nodes — WorkerNode instances that run in the coordinator's own process on asyncio queues. Required: a node_id (auto-generated if omitted). Optional: max_queue_depth (default 100) and concurrency (default 4).
• Remote nodes — RemoteWorkerNode instances that delegate over HTTP through an HttpWorkerTransport to a worker server on another machine.

Dispatch is least-loaded: every node exposes a load metric (active tasks plus queued tasks for local nodes; in-flight requests for remote ones), and submit() picks the healthy node with the lowest value across both pools. There is no separate scheduler process to deploy — the coordinator is just your Python program.

Step 1 — Start a worker

On each worker machine, export the token and start the server. The CLI route needs the [http] extra for uvicorn:

export REPLICATE_API_TOKEN=r8_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
replicate-agent workers start --host 0.0.0.0 --port 7999 --node-id gpu-node-1 --concurrency 8

Starting Worker Node
  Node ID:     gpu-node-1
  Address:     http://0.0.0.0:7999
  Concurrency: 8
  Token:       r8_x...xxxx
INFO:     Uvicorn running on http://0.0.0.0:7999 (Press CTRL+C to quit)

All flags shown are optional — --host 0.0.0.0, --port 7999, and --concurrency 8 are the defaults; only --node-id is worth setting explicitly (it defaults to worker-<pid>). The same server is available programmatically, which is how you would embed it in your own service:

import asyncio
from replicate_mcp.worker_server import serve_worker

asyncio.run(serve_worker(host="0.0.0.0", port=7999))

serve_worker() also accepts node_id, max_concurrency=8, enable_circuit_breaker=True, and an optional circuit_config. Either way the worker exposes three HTTP endpoints:

• POST /execute — run an agent invocation, return a TaskResult as JSON.
• GET /health — liveness probe; includes circuit-breaker state, and returns 503 when the circuit is OPEN.
• GET /metrics — load counters: active_tasks, total_processed, plus circuit metrics.

Step 2 — Health-check from the coordinator

Before writing coordinator code, prove the network path works from the coordinator machine. Raw curl shows you exactly what the transport will see:

curl http://gpu-node-1:7999/health

{"status": "healthy", "node_id": "gpu-node-1", "circuit": {"state": "closed",
 "failure_count": 0, "success_count": 0, "last_failure_at": null,
 "recovery_timeout": 60.0, "half_open_max_calls": 3, "half_open_calls": 0,
 "can_execute": true}}

The friendlier CLI wrapper does the same check plus a metrics read:

replicate-agent workers ping http://gpu-node-1:7999

✓ Worker at http://gpu-node-1:7999 is healthy
  Active tasks:     0
  Total processed:  0

✗ Worker at http://gpu-node-1:7999 is unreachable
# → check the worker bound 0.0.0.0 (not 127.0.0.1) and port 7999 is open

Step 3 — Build the coordinator

The coordinator combines remote and local nodes in one pool. Save as coordinator.py (this assumes the llama_chat agent from the Quickstart is registered on the workers):

import asyncio
from replicate_mcp.distributed import (
    DistributedExecutor,
    HttpWorkerTransport,
    RemoteWorkerNode,
    WorkerNode,
)

async def main():
    async with DistributedExecutor() as executor:
        # Remote GPU workers — HTTP transport, 120 s default request timeout
        for host in ("gpu-node-1", "gpu-node-2"):
            transport = HttpWorkerTransport(f"http://{host}:7999")
            executor.add_remote_node(RemoteWorkerNode(host, transport=transport))

        # Local in-process fallback node
        executor.add_node(WorkerNode("local-fallback", concurrency=4))

        handle = await executor.submit("llama_chat", {"prompt": "Say hello in 5 words"})
        result = await handle  # TaskResult

        print(result.node_id, result.status.value, f"{result.elapsed_ms:.0f}ms")
        for chunk in result.chunks:
            if chunk.get("done"):
                print(chunk.get("output"))

asyncio.run(main())

python coordinator.py

gpu-node-1 done 824ms
Hello there, nice to meet!

submit() returns a TaskHandle — a future you can hold while submitting more work, then await for the TaskResult. The result carries task_id, agent_name, node_id (which machine ran it), chunks (the streamed output), status (pending / running / done / failed), error, and elapsed_ms. For batches, await executor.run_many([("llama_chat", {...}), ...]) submits concurrently and returns results in input order; executor.stream(...) yields chunks from a single task as an async iterator.

Two failure modes are worth knowing by name: NodeOverloadError means a local node's queue hit max_queue_depth (the executor retries on another node up to max_retries=2 times before raising), and NoHealthyNodesError means the candidate pool is empty — every node is unhealthy or circuit-open.

Step 4 — Worker circuit breakers and failover

Since v0.8.0 each worker server tracks its own reliability with a WorkerCircuitBreaker (enabled by default; same CLOSED → OPEN → HALF_OPEN machine described in Resilience & caching, applied to the whole worker rather than one model). The state is serialized as a WorkerCircuitState — the circuit object you saw in the /health response — so coordinators can make routing decisions without any shared infrastructure:

• When the worker's circuit is OPEN, GET /health returns 503 and POST /execute rejects new tasks with 503.
• The coordinator checks circuit state before dispatch (RemoteWorkerNode.check_circuit_state()) and skips OPEN nodes entirely, raising WorkerCircuitOpenError internally and routing to the next healthy node.
• HALF_OPEN nodes stay in the pool but their load is multiplied by 1.5 during selection, so probe traffic trickles back instead of flooding a recovering worker.

To force a failover test, stop one worker (Ctrl+C on its host) and re-run coordinator.py: the printed node_id moves to a surviving node, and once nothing remote is healthy, to local-fallback.

Verification

Common mistakes

Next steps

• Troubleshooting — runbook entries for unreachable workers and open circuits.
• Plugins & observability — export per-node metrics and traces to your collector.
• Python API reference — full signatures for DistributedExecutor, WorkerNode, and the transports.