Drive OKMQ with AI agents — through MCP or a small worker script
The organising idea behind every pattern below: one queue per actor, tags for addressing. An actor is a model, an agent, or a worker role with its own queue as its inbox — qwen, gpt, opus, reviewer, impl. The dispatcher (your MCP session or a small worker script) decides who gets work and sends to that actor's queue. The actor polls its own queue, does the work, and sends its result onward by addressing another actor's queue. A tag attaches metadata to a message — a priority, a correlation id, a routing hint — but the actor's queue is where it lives.
This mirrors how crew-supervisor tools (e.g. firstmate) structure fleets: a liaison dispatches, each crewmate works from its own home, work moves between named actors. The queue is the home; the send is the routing.
Two consequences worth internalising: escalation is a resend — that carries the failure. When actor A fails a task, the dispatcher sends it to actor B's queue with the error appended (see the recipe in Start here). No retagging, no server magic — and the retry gets smarter because it knows what went wrong. And verification lives with the dispatcher — it owns the task context, so it owns the "was this done to satisfaction?" question. Actors report back; the dispatcher verifies (mechanically or with a judge) and routes accordingly.
One constraint to know up front: each queue is a channel, and the free tier has 1. But tags give you per-actor separation on that single queue: each actor polls its own tag, and receives never collide — two actors polling the same tag are handed different messages, atomically. okmq_status / okmq_peek accept tag filters for per-actor views. What you give up versus one-queue-per-actor: retry strategy, lock_duration, TTL, and auto_ack are queue-level settings, so every actor on a shared queue shares one policy — and a shared failed set instead of per-actor dead letters. Start tag-separated on the free tier; promote heavy actors to their own queues when you need per-actor policy.
Every pattern composes existing primitives — queues route work, locks claim it, retries recover it. No server changes needed. Pick one and prototype it in an afternoon.
https://api.okmq.net/mcp and it can create queues, send, receive, and ack in plain English. Zero code, good for interactive experiments and coordination between agents in different tools.Both share the same API key, queues, and tags — an MCP agent session and a worker script can cooperate on the same queues without knowing about each other.
N agents of the same kind poll one shared queue with the same tag filter — the one place tags act as capabilities rather than addressing. lock_duration_seconds is the claim lease: an agent that crashes mid-task releases the message to the next poller. The queue's retry strategy absorbs flaky models. No coordinator process; a heterogeneous fleet out of the box. Runs on the free tier's single channel — and this is the shape every tag-separated actor setup takes (see the constraint note above). When you need per-agent isolation instead — separate lock budgets, retry policies, per-agent dead letters — promote the tag to a queue: one queue per actor, same semantics.
Cost optimization across actor queues. The dispatcher sends the task to the cheap actor's queue (worker-a); when the result fails verification — loud (nack, lock expiry) or quiet (verifier rejects) — it resends to the next rung with the failure appended: worker-b, then opus. Each rung is a more expensive (usually more capable) model, and each retry is smarter than the last because it carries the error. The hard part is not the routing; it is knowing the task failed. Three verifier tiers, cheapest first: mechanical checks (compile, tests, lint — binary, scriptable), a judge agent that reads the result against the task (same trust model as pattern 6), and a human as the terminal rung — a message that exhausts max_delivery_attempts lands in status=failed, and that failed set is the human review inbox. The ladder routes on detectable failure; the verifier decides what is detectable. The end-to-end recipe in Start here implements exactly this ladder.
The implement actor finishes and sends tagged reviewer to the reviewer actor's queue. The reviewer reads, sends comments tagged back to impl-{taskId}. Retry strategy redelivers until the implementer picks it up. A conversation over queues, no controller process — tags are correlation ids, queues are mailboxes. Any MCP-capable agent is a node for free, and a Claude Code session and an opencode session can be the two ends.
One task decomposes into N subtasks; the results must be reassembled. The dispatcher sends one message per subtask to the workers' queue, each tagged with the parent id (job-{id}), then polls its own results queue for the N replies. Workers send results tagged back to dispatcher. When N results are in, the dispatcher synthesizes and acks the parent. The primitives make it durable: tags carry the correlation id, the queues hold partial results while the rest still run, and no coordinator needs to remember anything across a crash. Throughput version of pattern 6 — instead of racing two models on the same task, you parallelize different subtasks.
The single most requested agent pattern: pause before anything irreversible. When an actor reaches a gated step — deploy, send an email, delete a branch — it does not act. It sends the proposed action to an approval queue, then nacks its own task so the gate window is bounded by redelivery: set retry_after_seconds on the nack (or give the queue a long fixed retry) large enough that the human responds before max_delivery_attempts runs out — under defaults (strategy none, 3 attempts) an unattended gate exhausts the task into status=failed on the first expiry. Alternatively leave the task locked for the lock_duration you trust. The human is the verifier, from any MCP session: "what's waiting for approval?" (okmq_peek), then "okmq_ack it" to grant or "okmq_nack it back to impl with the objection in the body" to reject. On grant, the actor resumes; on reject, the actor retries with the objection appended. This is the constructive counterpart to pattern 2's terminal human rung: there the human reviews failures; here the human grants permission before the work happens.
Send the same hard task to two actors' queues and let a judge actor pick the winner. Two times the tokens, but it buys latency on ambiguous tasks. The judge reads both replies (tags carry the correlation id), acks the winner, and nacks the loser.
okmq_status per actor queue gives pending/processing/failed counts per actor — the fleet view is one status call per actor. okmq_peek with status failed is dead-letter inspection per actor. A small worker script plus a terminal UI — or just ask your MCP agent "what's stuck?"
Queues route between actors; everything else shapes behavior:
delivery_time + tag = per-actor cron — a message scheduled in the future appears when that actor's shift startslock_duration_seconds = how long an actor may hold a task before it is fair gamep0 / p1 tags in one actor's queue — the actor polls p0 first, falls through to p1 when empty (with per-priority model assignment as a bonus)Agent task times are bimodal and hard to predict: the same task can take 40 seconds or 40 minutes depending on the model's path. lock_duration_seconds is a fixed lease, and getting it wrong fails silently in both directions: too short and the lease expires while the actor still works, so another poller pops the same message and runs it concurrently — and note the sharp edge: a pop does not increment delivery_attempt (only nack and abort-requeue do), so a chronically slow actor can have its task stolen repeatedly without ever surfacing in counts, silent duplicate work. Too long and a crashed actor's task sits invisible in status=processing until the lease expires.
Three mitigations make the existing primitives safe for unpredictable agents:
lock_duration (the production worker pairs a 900s task timeout with a 900s lock — use 840/900). Deliberate overruns die at the timeout before the lock lapses; only genuine crashes release the message early.If agents routinely outgrow their leases, that is the signal to widen the lease or split the task into checkpointed stages — or want a lock-extension primitive (a heartbeat that pushes locked_until forward without redelivery). It does not exist yet; if you find yourself tuning lock_duration per task length, that is the missing primitive.
okmq_status calls (one per actor queue) instead of one tag-filtered call.okmq_queues_list-style tool would let agents discover each other.The simplest end-to-end prototype is pattern 2 at two rungs: one queue per actor (worker-a, worker-b), one dispatcher, one worker script per actor. The dispatcher sends a task; when verification fails, it resends to the next actor with the failure appended. This is the whole game — every pattern above is a variation on it.
The worker loop — receive → process → ack/nack. The full TypeScript version (~15 lines, same imports as below) and a bash daemon variant that shells out to a headless agent are in the MCP guide's Unattended Worker Pattern:
import { client, getQueuesByQueueMessages, postQueuesByQueueAck } from './client'
client.setConfig({ headers: { Authorization: process.env.TOKEN } })
const queue = process.env.OKMQ_QUEUE || 'worker-a'
while (true) {
const { data: messages } = await getQueuesByQueueMessages({ path: { queue }, query: { limit: 1 } })
if (!messages?.length) { await new Promise(r => setTimeout(r, 2000)); continue }
const msg = messages[0]
try {
await doWork(msg.body) // your verifier runs here too:
await postQueuesByQueueAck({ // mechanical checks, or send to a judge
path: { queue },
body: [{ id: msg.id, ack: true }],
})
} catch (error) {
await postQueuesByQueueAck({
path: { queue },
body: [{ id: msg.id, ack: false }], // nack: retry strategy redelivers it
})
}
}The dispatcher — resend on failure, carrying the error. This is the ladder:
const task = { id: 't1', body: 'Refactor auth.ts to use the new middleware API' }
await postQueuesByQueueMessages({ path: { queue: 'worker-a' }, body: [task] })
// ... later, after the verifier rejects worker-a's output:
await postQueuesByQueueMessages({
path: { queue: 'worker-b' },
body: [{ id: 't1-b', body: task.body + '\n\nPrevious attempt failed: ' + verifierError + '\nFix and try again.' }],
}Send the first task from any MCP session:
{
"name": "okmq_send",
"arguments": {
"queue": "worker-a",
"messages": [
{ "id": "t1", "body": "Run the test suite and summarize failures." }
]
}
}Then watch the ladder move: okmq_status on worker-a, then on worker-b after a failed verification.