Background Jobs for Long-Running MCP Tools — BullMQ, pg-boss, and progress notifications

When a synchronous tool call is not enough

MCP tool calls are request/response: the AI agent sends a tools/call request and waits for the response. The MCP specification imposes no hard timeout on tool calls, but practical limits exist at every layer:

• The AI agent's LLM context has a limited timeout for tool execution before it marks the call as failed
• HTTP proxies and load balancers commonly have 30–120 second read timeouts
• The MCP server process may have per-request timeouts configured
• Holding an open connection and a database pool slot for 60+ seconds blocks other tool calls

Operations that regularly exceed 10–15 seconds should be delegated to background jobs. Common examples:

The async job pattern for MCP tools

The pattern has three components: an enqueue tool, a status resource, and a worker process.

Step 1: The enqueue tool

The tool handler enqueues the work and returns immediately with a job ID. The agent receives the ID and can track the job without holding an open connection.

import { Queue } from 'bullmq';
import { Redis } from 'ioredis';
import { z } from 'zod';

const redis = new Redis(process.env.REDIS_URL);
const exportQueue = new Queue('export', { connection: redis });

const ExportSchema = z.object({
  format: z.enum(['csv', 'json', 'parquet']),
  date_from: z.string().datetime(),
  date_to: z.string().datetime(),
  idempotency_key: z.string().optional(),
});

server.tool(
  'export_data',
  ExportSchema.shape,
  async (args) => {
    const parsed = ExportSchema.parse(args);

    // Use idempotency_key as jobId — agent retries don't create duplicate exports
    const jobId = parsed.idempotency_key ?? crypto.randomUUID();

    await exportQueue.add('export', parsed, {
      jobId,                           // deduplicate by this ID
      removeOnComplete: { age: 3600 }, // keep completed jobs for 1 hour
      removeOnFail: { age: 86400 },    // keep failed jobs for 24 hours
    });

    return {
      content: [{
        type: 'text',
        text: JSON.stringify({
          job_id: jobId,
          status: 'queued',
          poll_resource: `job:${jobId}`,
          estimated_seconds: 30,
        }),
      }],
    };
  }
);

Step 2: The status resource

Expose job status via the MCP resources protocol. The agent calls resources/read with the URI returned by the tool to poll for completion.

// Register a resource template for job status
server.resource(
  'job-status',
  new ResourceTemplate('job:{jobId}', { list: undefined }),
  async (uri, { jobId }) => {
    const job = await exportQueue.getJob(jobId);

    if (!job) {
      return {
        contents: [{ uri: uri.href, mimeType: 'application/json', text: JSON.stringify({ status: 'not_found' }) }],
      };
    }

    const state = await job.getState();  // 'waiting' | 'active' | 'completed' | 'failed'

    const payload: Record = {
      job_id: jobId,
      status: state,
      progress: job.progress,
    };

    if (state === 'completed') {
      payload.result = job.returnvalue;
      payload.completed_at = new Date(job.finishedOn!).toISOString();
    }

    if (state === 'failed') {
      payload.error = job.failedReason;
      payload.failed_at = new Date(job.finishedOn!).toISOString();
    }

    return {
      contents: [{
        uri: uri.href,
        mimeType: 'application/json',
        text: JSON.stringify(payload),
      }],
    };
  }
);

Step 3: The worker process

The worker runs in a separate process (or a separate instance of the same process). Separating workers from the MCP server means a slow job does not block tool call handling and the worker can be scaled independently.

// worker.ts — separate process
import { Worker } from 'bullmq';

const worker = new Worker('export', async (job) => {
  const { format, date_from, date_to } = job.data;
  const rows = [];
  let page = 0;

  while (true) {
    const batch = await db.query(
      'SELECT * FROM events WHERE created_at BETWEEN $1 AND $2 LIMIT 1000 OFFSET $3',
      [date_from, date_to, page * 1000]
    );
    if (batch.rows.length === 0) break;

    rows.push(...batch.rows);

    // Report progress — AI agent can read this via resources/read
    await job.updateProgress(Math.round((rows.length / estimatedTotal) * 100));
    page++;
  }

  const output = formatData(rows, format);
  const url = await uploadToStorage(output, `exports/${job.id}.${format}`);
  return { download_url: url, row_count: rows.length };
}, { connection: redis, concurrency: 3 });

worker.on('failed', (job, err) => {
  console.error(JSON.stringify({ event: 'job_failed', jobId: job?.id, error: err.message }));
});

Progress notifications via the MCP protocol

For AI agents that need live updates (not just final results), MCP supports notifications/progress — the server pushes incremental progress events over the SSE connection while the tool call is running. The agent receives updates without polling.

Progress notifications require the agent to include a _meta.progressToken in the tool call. The MCP server sends notifications referencing that token.

// Tool handler with progress notifications
server.setRequestHandler(CallToolRequestSchema, async (request, context) => {
  if (request.params.name !== 'export_data') return;

  const progressToken = request.params._meta?.progressToken;

  // For jobs that run in-process (not via a separate worker):
  const sendProgress = progressToken
    ? (progress: number, total: number) => {
        context.sendNotification({
          method: 'notifications/progress',
          params: { progressToken, progress, total },
        });
      }
    : () => {};

  // Long-running operation with progress reporting
  const rows = [];
  const batches = await countBatches(request.params.arguments);

  for (let i = 0; i < batches; i++) {
    const batch = await fetchBatch(i, request.params.arguments);
    rows.push(...batch);
    sendProgress(i + 1, batches);
  }

  return {
    content: [{ type: 'text', text: JSON.stringify({ row_count: rows.length, data: rows }) }],
  };
});

Progress notifications require the Streamable HTTP or SSE transport — stdio transport does not support server-initiated notifications mid-request. For stdio-transport MCP servers, fall back to the resource polling pattern.

pg-boss: background jobs with PostgreSQL only

If your MCP server already uses PostgreSQL and you want to avoid adding Redis as a dependency, pg-boss provides a PostgreSQL-backed job queue with comparable features: job deduplication, retries, TTL, and concurrency control.

import PgBoss from 'pg-boss';

const boss = new PgBoss(process.env.DATABASE_URL);
await boss.start();

// Enqueue a job (identical interface to BullMQ from the tool handler's perspective)
const jobId = await boss.send('export', jobData, {
  id: idempotencyKey,          // deduplication key
  retryLimit: 3,
  retryDelay: 30,
  expireInSeconds: 3600,
});

// Worker (can run in same process or separate)
await boss.work('export', { teamSize: 3 }, async (job) => {
  // Process job.data
  return { download_url: url, row_count: count };
});

// Poll job status
const job = await boss.getJobById(jobId);
// job.state: 'created' | 'active' | 'completed' | 'failed' | 'cancelled'
// job.output: return value when completed

pg-boss stores jobs in a pgboss schema in your PostgreSQL database. The job table is indexed for efficient polling. The tradeoff vs BullMQ: PostgreSQL-backed queues have higher per-job latency (1–10ms vs Redis's sub-millisecond) but eliminate Redis as an operational dependency. For jobs that run over seconds, the queue latency difference is irrelevant.

Monitoring the worker pipeline with AliveMCP

The MCP protocol probe (initialize → tools/list) confirms the MCP server is running. It does not confirm the worker process is running, the Redis/PostgreSQL queue is draining, or that jobs complete within expected windows. A worker process crash is invisible to the protocol probe.

The two monitoring patterns for background job pipelines:

1. Canary job tool call

Add a health_check_job tool that enqueues a sentinel job and polls for completion:

// health_check_job tool — validates the full worker pipeline
server.tool('health_check_job', {}, async () => {
  const jobId = `health-${Date.now()}`;
  await healthQueue.add('sentinel', { sentinel: true }, { jobId, priority: 1 });

  // Poll for up to 30 seconds
  const deadline = Date.now() + 30_000;
  while (Date.now() < deadline) {
    await new Promise(r => setTimeout(r, 1000));
    const job = await healthQueue.getJob(jobId);
    if (!job) continue;
    const state = await job.getState();
    if (state === 'completed') {
      return { content: [{ type: 'text', text: JSON.stringify({ ok: true, queue_healthy: true }) }] };
    }
    if (state === 'failed') {
      throw new Error(`sentinel_job_failed: ${job.failedReason}`);
    }
  }
  throw new Error('sentinel_job_timeout: worker not processing within 30s');
});

Configure AliveMCP to call health_check_job on each probe cycle. If the canary job doesn't complete within 30 seconds, AliveMCP alerts — the worker process may have crashed, Redis may be unreachable, or the queue may be backlogged.

2. Health endpoint with queue depth

// Queue health in /health endpoint
app.get('/health', async (req, res) => {
  const counts = await exportQueue.getJobCounts('waiting', 'active', 'failed');

  if (counts.failed > 50) {
    return res.status(503).json({
      status: 'degraded',
      reason: 'high_failure_rate',
      queue: counts,
    });
  }

  if (counts.waiting > 200) {
    return res.status(503).json({
      status: 'degraded',
      reason: 'queue_backlog',
      queue: counts,
    });
  }

  res.json({ status: 'ok', queue: counts });
});

Frequently asked questions

Further reading

• Redis Patterns for MCP Servers — caching, pub/sub, and session state
• MCP Progress Notifications — streaming job status to the AI agent
• MCP Server Idempotency — preventing duplicate side effects across agent retries
• MCP Resources API — exposing structured data for AI agent consumption
• Worker Threads for MCP Servers — CPU-intensive tools without blocking the event loop
• Graceful Shutdown — draining in-flight jobs before process exit