Operate workers

An AgentWorker consumes jobs as soon as it is constructed. Registry keys must match names submitted by the producer. Missing entries fail the job with a named registry error.

Start a worker

Use the dependencies from the quickstart, plus npm install openai. Set OPENAI_API_KEY, save this as worker.ts, and run npx tsx worker.ts. This worker matches the producer example.

worker.ts

import { Agent, openai } from "@agentium/core";
import { AgentWorker } from "@agentium/queue";

const assistant = new Agent({
  name: "assistant",
  model: openai(process.env.OPENAI_MODEL ?? "gpt-6.1-sol"),
  instructions: "Summarize the supplied content concisely.",
  register: false,
});
const worker = new AgentWorker({
  connection: process.env.REDIS_URL ?? "redis://127.0.0.1:6379",
  queueName: "agentium-jobs",
  agentRegistry: { assistant },
  concurrency: 5,
});
console.log("Worker is listening for jobs.");

let stopping = false;
async function shutdown() {
  if (stopping) return;
  stopping = true;
  try {
    await worker.stop(30_000);
    await assistant.close();
  } catch (error) {
    // A drain timeout does not prove active jobs have stopped using the agent.
    console.error("Worker did not drain cleanly:", error);
    process.exitCode = 1;
  }
}
process.once("SIGINT", () => { void shutdown(); });
process.once("SIGTERM", () => { void shutdown(); });

start() remains available for compatibility but performs no work. concurrency defaults to 5. Set it deliberately according to provider quotas, tool limits, and available resources.

Register other work

Use teamRegistry for team jobs and workflowRegistry for workflow jobs. agentRegistry is required even for a worker that only handles other job types; pass {} in that case. See team jobs and the WorkerConfig reference. The worker forwards the submitted session, user, and tenant fields to the run. Enqueue only host-authorized identity values. In-process Agent objects can serve concurrent jobs, so external stores and application callbacks must support the concurrency you choose.

Understand progress

For agent jobs, the worker listens to run.stream.chunk and increments a counter scoped to that run ID. It removes the listener when execution finishes and waits for outstanding progress writes. This counter does not estimate percent complete; a non-streaming model call may not produce chunk events. bridgeEventBusToJob(eventBus, job, runId) is available for custom BullMQ workers. It counts matching chunks and logs matching tool.call events, then returns a cleanup function. Call that function in finally. The helper does not log tool.result, and the built-in worker uses its own progress listener rather than this helper.

Drain before closing dependencies

await worker.stop(timeoutMs) asks BullMQ to close the worker and waits for active jobs, with a default 30-second deadline. A timeout rejects with Worker drain timeout; it does not forcibly abort a provider or reverse side effects. Close Agents and their owned storage only after work has drained. Configure your process manager’s termination deadline to allow that sequence.

Scale across processes

Workers with the same Redis connection and queue name share deliveries. Three workers at concurrency 5 allow up to 15 active jobs. Producer retries still apply to failed executions; worker options do not configure retry policy. For crash recovery with persisted execution state, use durable delivery.

Build

Agent applications

Harness runtime

Teams and workflows

Data and capabilities

Interfaces and jobs

Voice and telephony

Verify and manage

Start a worker

Register other work

Understand progress

Drain before closing dependencies

Scale across processes

​Start a worker

​Register other work

​Understand progress

​Drain before closing dependencies

​Scale across processes

Start a worker

Register other work

Understand progress

Drain before closing dependencies

Scale across processes