MCP Tools for QStash — HTTP job queue, CRON schedules, callbacks, retries, deduplication

QStash vs traditional message queues

QStash is an HTTP-based message queue — you publish a message to QStash with a destination URL, and QStash delivers it via HTTP POST to that URL. Unlike Redis-backed queues (BullMQ, Celery) or AMQP brokers (Azure Service Bus, RabbitMQ), QStash requires no persistent connection, no worker process polling a queue, and no broker infrastructure in your deployment. This makes it ideal for serverless MCP servers (Vercel, Netlify, Cloudflare Workers) that cannot maintain long-lived connections. The destination endpoint can be any publicly reachable URL — the MCP server, a webhook handler, or a serverless function.

import { Client, Receiver } from '@upstash/qstash';

const qstash = new Client({
  token: process.env.QSTASH_TOKEN!,
});

// The receiver is used to verify incoming QStash delivery requests
const receiver = new Receiver({
  currentSigningKey: process.env.QSTASH_CURRENT_SIGNING_KEY!,
  nextSigningKey: process.env.QSTASH_NEXT_SIGNING_KEY!,
  // nextSigningKey handles the key rotation window — always include both
});

// Publish a message to an endpoint
async function dispatchTask(taskData: object): Promise<{ messageId: string }> {
  const response = await qstash.publishJSON({
    url: `${process.env.WORKER_BASE_URL}/api/tasks/process`,
    body: taskData,
    retries: 3,           // retry 3 times on non-2xx response
    timeout: 30,          // seconds — QStash waits this long for your endpoint to respond
    // Callback URL receives the final delivery result (success or failure)
    // callback: `${process.env.WORKER_BASE_URL}/api/tasks/callback`,
    // failureCallback: `${process.env.WORKER_BASE_URL}/api/tasks/failure`,
  });
  return { messageId: response.messageId };
}

QStash operates as a delivery guarantee layer — it handles retries, backoff, and scheduling, while your endpoint handles the actual work. The endpoint must be publicly reachable by QStash's delivery infrastructure. For local development, use ngrok or the QStash local development server (npx @upstash/qstash-cli dev) to receive deliveries on localhost.

Signature verification — required for all receiving endpoints

QStash signs all delivery requests with an HMAC-SHA256 signature in the Upstash-Signature header. Verifying this signature is mandatory — without it, anyone who discovers your endpoint URL can trigger task execution by sending a crafted POST request. QStash uses two signing keys (current and next) to support zero-downtime key rotation; the Receiver class tries both keys automatically.

import { NextRequest, NextResponse } from 'next/server';

// Next.js App Router example — same pattern for Express, Hono, etc.
export async function POST(req: NextRequest): Promise {
  // Read raw body BEFORE parsing — signature covers the raw bytes
  const rawBody = await req.text();
  const signature = req.headers.get('upstash-signature');

  if (!signature) {
    return NextResponse.json({ error: 'Missing signature' }, { status: 401 });
  }

  // Verify signature — throws if invalid
  try {
    await receiver.verify({
      signature,
      body: rawBody,
      clockTolerance: 5,  // seconds of clock skew to allow
    });
  } catch (err) {
    return NextResponse.json({ error: 'Invalid signature' }, { status: 401 });
  }

  // Parse body only after verification
  let taskData: unknown;
  try {
    taskData = JSON.parse(rawBody);
  } catch {
    return NextResponse.json({ error: 'Invalid JSON' }, { status: 400 });
  }

  // Process the task
  try {
    await processTask(taskData);
    // Return 2xx to signal successful delivery — QStash will not retry
    return NextResponse.json({ status: 'processed' });
  } catch (err) {
    // Return non-2xx to trigger retry
    return NextResponse.json({ error: 'Processing failed' }, { status: 500 });
  }
}

// Express example
async function qstashHandler(req: express.Request, res: express.Response) {
  // Use express.raw() middleware to get raw body: app.use('/api/qstash', express.raw({ type: '*/*' }))
  const rawBody = req.body.toString('utf-8');
  const signature = req.headers['upstash-signature'] as string;

  try {
    await receiver.verify({ signature, body: rawBody });
  } catch {
    return res.status(401).json({ error: 'Invalid QStash signature' });
  }

  const data = JSON.parse(rawBody);
  await processTask(data);
  res.json({ status: 'ok' });
}

Always use express.raw() (not express.json()) for QStash endpoints — the signature is computed over the raw request body, and JSON parsing before verification may re-serialise the body with different whitespace, invalidating the signature check. Read the raw bytes first, verify, then parse.

Delayed delivery, deduplication, and queues

QStash supports delivery delay (message is held and delivered after a delay), deduplication (prevent duplicate deliveries within a 24-hour window), and queues (named FIFO delivery channels with rate limits). Queues are distinct from topics — a QStash queue delivers messages sequentially to one destination URL at a configurable rate, useful for rate-limiting calls to APIs that have per-second limits.

// Delayed delivery — message held for 60 seconds before first delivery attempt
await qstash.publishJSON({
  url: endpointUrl,
  body: { userId: 'usr_123', action: 'send-welcome-series' },
  delay: '1m',  // '30s', '5m', '1h', '2d' — or integer seconds
});

// Deduplication — same deduplicationId within 24 hours = single delivery
async function dispatchOnce(taskId: string, data: object): Promise {
  const response = await qstash.publishJSON({
    url: endpointUrl,
    body: data,
    deduplicationId: taskId,  // deterministic ID — second publish returns same messageId
  });
  return response.messageId;
}

// Named queue with rate limit — useful for rate-limited API calls
const q = qstash.queue({ queueName: 'openai-requests' });
await q.upsert({ parallelism: 1 });  // sequential delivery (1 concurrent request)

await q.enqueueJSON({
  url: endpointUrl,
  body: { prompt: 'analyze this document', docId: 'doc_456' },
});

// Batch publish — multiple messages in one API call
await qstash.batchJSON([
  { url: endpointUrl, body: { item: 'a' } },
  { url: endpointUrl, body: { item: 'b' } },
  { url: endpointUrl, body: { item: 'c' }, delay: '30s' },
]);

QStash queues support a parallelism setting that limits concurrent in-flight deliveries. Setting parallelism: 1 creates a strict FIFO queue — the next message is not delivered until the previous one receives a 2xx response. This is useful for MCP tools that must call APIs with per-request rate limits or that process mutations that must not run concurrently for the same resource.

CRON schedules — recurring tasks from MCP tools

QStash schedules send recurring HTTP deliveries on a cron expression. Unlike cron jobs running on your server, QStash schedules are managed externally and survive server restarts. MCP tools can create, list, pause, and delete schedules programmatically — enabling agents to set up and manage recurring workflows.

// Create a CRON schedule — fires every Monday at 9 AM UTC
const schedule = await qstash.schedules.create({
  destination: `${process.env.WORKER_BASE_URL}/api/tasks/weekly-report`,
  cron: '0 9 * * 1',         // cron expression (UTC timezone)
  body: JSON.stringify({ reportType: 'weekly-summary' }),
  headers: { 'Content-Type': 'application/json' },
  retries: 2,
});
console.log(`Schedule created: ${schedule.scheduleId}`);

// List all schedules for this QStash account
const schedules = await qstash.schedules.list();
for (const s of schedules) {
  console.log(`Schedule ${s.scheduleId}: ${s.cron} → ${s.destination}`);
}

// Pause a schedule (stops future deliveries without deleting)
await qstash.schedules.pause({ scheduleId: schedule.scheduleId });

// Resume a paused schedule
await qstash.schedules.resume({ scheduleId: schedule.scheduleId });

// Delete a schedule — agent can clean up its own scheduled tasks
await qstash.schedules.delete(schedule.scheduleId);

// MCP tool: let agents create scheduled tasks
function mcpCreateScheduleTool(cron: string, taskData: object): Promise<{ scheduleId: string }> {
  return qstash.schedules.create({
    destination: `${process.env.WORKER_BASE_URL}/api/tasks/scheduled`,
    cron,
    body: JSON.stringify(taskData),
    headers: { 'Content-Type': 'application/json' },
  }).then(s => ({ scheduleId: s.scheduleId }));
}

QStash schedule deliveries go through the same signature verification and retry mechanism as regular publishes. The receiving endpoint cannot distinguish a scheduled delivery from a manually published one — both carry the same Upstash-Signature header. Use a custom header in the schedule's headers (e.g., X-Trigger: schedule) if your endpoint needs to behave differently for scheduled vs on-demand invocations.

Failure callbacks and message status

QStash can call a failure callback URL when all retry attempts for a message are exhausted. This replaces the dead-letter queue pattern — the failure callback receives the original message body plus delivery failure details. Configure failureCallback on every publish to prevent silent data loss when downstream endpoints are persistently failing.

// Publish with failure callback
async function dispatchWithFailureHandling(data: object): Promise<{ messageId: string }> {
  const response = await qstash.publishJSON({
    url: `${process.env.WORKER_BASE_URL}/api/tasks/process`,
    body: data,
    retries: 3,
    failureCallback: `${process.env.WORKER_BASE_URL}/api/tasks/dead-letter`,
    // callback fires on EVERY delivery attempt (success or failure)
    // callback: `${process.env.WORKER_BASE_URL}/api/tasks/status`,
  });
  return { messageId: response.messageId };
}

// Failure callback endpoint — receives delivery failure info
// Verify signature here too (QStash signs callback requests with the same key)
async function handleFailureCallback(rawBody: string, signature: string): Promise {
  await receiver.verify({ signature, body: rawBody });
  const payload = JSON.parse(rawBody);
  // payload.sourceMessageId — original message ID that failed
  // payload.topicName / payload.url — destination that failed
  // Save to your own dead-letter store
  await storeDeadLetter({
    messageId: payload.sourceMessageId,
    failedAt: new Date(),
    originalBody: payload.body,
  });
}

// Check message delivery status
async function getMessageStatus(messageId: string): Promise