@agentium/queue

npm install @agentium/queue bullmq ioredis

import { AgentQueue } from "@agentium/queue";

new AgentQueue(config: QueueConfig)

async enqueueAgentRun(opts: {
  agentName: string;
  input: string;
  sessionId?: string;
  userId?: string;
  priority?: number;
  delay?: number;
}): Promise<{ jobId: string }>

async enqueueWorkflow(opts: {
  workflowName: string;
  initialState?: Record<string, unknown>;
  sessionId?: string;
  priority?: number;
  delay?: number;
}): Promise<{ jobId: string }>

async getJobStatus(jobId: string): Promise<JobStatus>

async cancelJob(jobId: string): Promise<void>

queue.onCompleted((jobId: string, result: RunOutput) => {
  console.log(`Job ${jobId} completed:`, result.text);
});

queue.onFailed((jobId: string, error: Error) => {
  console.error(`Job ${jobId} failed:`, error.message);
});

async close(): Promise<void>

import { AgentWorker } from "@agentium/queue";

new AgentWorker(config: WorkerConfig)

start(): void

async stop(): Promise<void>

type JobPayload = AgentJobPayload | WorkflowJobPayload;

interface AgentJobPayload {
  type: "agent";
  agentName: string;
  input: string;
  sessionId?: string;
  userId?: string;
}

interface WorkflowJobPayload {
  type: "workflow";
  workflowName: string;
  initialState?: Record<string, unknown>;
  sessionId?: string;
}

interface JobStatus {
  jobId: string;
  state: "waiting" | "active" | "completed" | "failed" | "delayed";
  progress?: number;
  result?: RunOutput;
  error?: string;
  createdAt: Date;
  processedAt?: Date;
  finishedAt?: Date;
}

import { bridgeEventBusToJob } from "@agentium/queue";

const cleanup = bridgeEventBusToJob(eventBus, job, runId);
// Returns cleanup function to remove listeners

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

const agent = new Agent({
  name: "processor",
  model: openai("gpt-4o"),
  instructions: "Process the given data.",
});

// Producer
const queue = new AgentQueue({
  connection: { host: "localhost", port: 6379 },
});

const { jobId } = await queue.enqueueAgentRun({
  agentName: "processor",
  input: "Analyze this dataset",
});

console.log("Enqueued job:", jobId);

// Worker (can be in a separate process)
const worker = new AgentWorker({
  connection: { host: "localhost", port: 6379 },
  agentRegistry: { processor: agent },
  concurrency: 3,
});

worker.start();

// Monitor
queue.onCompleted((id, result) => {
  console.log(`Job ${id}:`, result.text);
});