Asynchronous jobs: queues, workers, retries (retry)

1. Why asynchronous jobs in ChatGPT App

In a perfect world, any MCP tool of yours would finish in a couple hundred milliseconds. But in real life, the interesting stuff is long and heavy:

• parsing a large CSV with a user’s purchase history;
• aggregating data from multiple external APIs, each of which might nap or respond with 503;
• building complex recommendations with lots of intermediate steps;
• generating large reports and presentations.

If you try to cram this into a single synchronous tool call, you’ll hit three problems.

First, timeouts. The ChatGPT session, HTTP infrastructure, and MCP client are not designed for “come back in five minutes” responses. A server that keeps a connection open too long will look “hung” to both ChatGPT and the user.

Second, load management. If a hundred users simultaneously launch a “super holiday gift analysis,” you don’t want your MCP server to hold a hundred long tasks synchronously within HTTP threads. You need a layer that can absorb spikes, distribute tasks across a queue, and process them with multiple workers.

Third, UX. A user clicks a button in the GiftGenius widget and stares at a single spinner for 40 seconds — it feels like old online banking. A “fast response + progress + ability to cancel” model is much nicer.

These problems are solved by a common pattern: “start → queue → background → events.”

2. Basic async-job architecture in the context of MCP

Let’s take our GiftGenius. Suppose we’ve added a new heavy scenario: “Deep analysis of preferences based on a friend’s purchase history and social media.” This can take several minutes, so:

1. The MCP tool accepts parameters from the model.
2. Instead of doing all the work right away, it creates a Job record in the database.
3. It enqueues the task.
4. It immediately replies to ChatGPT: “Analysis started, here is the jobId.”
5. A background worker takes the task from the queue, performs the heavy work, and along the way emits MCP events job.progress and job.partial, and at the end — job.completed or job.failed.

Architecturally, it looks roughly like this:

flowchart LR
    subgraph ChatGPT
      U[User] --> GPT[Model + ChatGPT UI]
    end

    GPT -->|call_tool analyze_preferences| MCP[MCP server]

    subgraph Backend
      MCP -->|create Job| DB[(Jobs DB)]
      MCP -->|enqueue| Q[Queue]
      W[Worker] -->|take job| Q
      W -->|update status/progress| DB
      W -->|MCP events: job.progress/job.completed| MCP
    end

    MCP -->|SSE events| GPT

The important idea: the MCP server is not necessarily a monolith. Often it acts as a façade over your internal async infrastructure: it accepts tool calls, creates jobs, and emits events, while separate worker processes do the heavy lifting.

3. Data model for an asynchronous job

Let’s start with a simple Job model. We’ll use TypeScript and a notional Node/MCP server so you can see how it fits your stack.

The simplest in-memory/DB model might look like this:

// openai/jobs/model.ts
export type JobStatus =
  | 'pending'
  | 'in_progress'
  | 'completed'
  | 'failed'
  | 'canceled';

export interface GiftJob {
  id: string;                // jobId
  type: 'deep_gift_analysis';
  status: JobStatus;
  payload: {
    recipientProfile: string;  // profile text/ID
    budget: number;
  };
  result?: unknown;          // final recommendations
  error?: string;            // error reason
  attempts: number;          // how many times we tried to execute
  createdAt: Date;
  updatedAt: Date;
}

In a real project, you’ll store GiftJob in Postgres, DynamoDB, Firestore, or elsewhere, but for the lecture the key fields are:

• status — current task state, reflected in both events and UX;
• attempts — counter for retry;
• error — for logs and debugging;
• payload — input data used by the worker to process the task.

4. The MCP tool that creates an async job

Imagine a tool named start_deep_analysis. Previously it might have done everything synchronously; now it only enqueues a task and returns a jobId.

// openai/tools/startDeepAnalysis.ts
import { v4 as uuid } from 'uuid';
import { createJobAndEnqueue } from '../jobs/queue';

// Pseudo-types for the MCP SDK
type StartDeepAnalysisInput = {
  recipientProfile: string;
  budget: number;
};

type StartDeepAnalysisOutput = {
  jobId: string;
  message: string;
};

export async function startDeepAnalysisTool(
  input: StartDeepAnalysisInput
): Promise<StartDeepAnalysisOutput> {
  const jobId = uuid();

  await createJobAndEnqueue({
    id: jobId,
    type: 'deep_gift_analysis',
    status: 'pending',
    payload: {
      recipientProfile: input.recipientProfile,
      budget: input.budget,
    },
    attempts: 0,
    createdAt: new Date(),
    updatedAt: new Date(),
  });

  return {
    jobId,
    message: `Started the deep analysis. Job ID: ${jobId}. I will send updates as they are ready.`,
  };
}

Key points:

• The MCP tool runs quickly: at most a couple of DB/queue calls;
• it returns a structured response with a jobId that ChatGPT can use in its “explanation to the user,” and that the GiftGenius widget can store in its widgetState.

Your JSON Schema for this tool simply describes jobId as a string and message as human-readable text — the model will understand that this is a task identifier and can reference it in subsequent steps of the conversation.

5. A simple queue and worker: training version

To avoid pulling in Redis, RabbitMQ, and everything else right now, we’ll make a simplified in-memory queue. In real production, of course, this would be a separate service (SQS/BullMQ/Cloud Tasks, etc.), but the logic will be the same.

First, a queue stub:

// openai/jobs/queue.ts
import type { GiftJob } from './model';

const jobs = new Map<string, GiftJob>();   // "DB" in memory
export const queue: string[] = [];         // simplified queue by id

export async function createJobAndEnqueue(job: GiftJob) {
  jobs.set(job.id, job);
  queue.push(job.id);
}

export function getJob(id: string): GiftJob | undefined {
  return jobs.get(id);
}

export function updateJob(id: string, patch: Partial<GiftJob>) {
  const job = jobs.get(id);
  if (!job) return;
  const updated: GiftJob = { ...job, ...patch, updatedAt: new Date() };
  jobs.set(id, updated);
}

Now a primitive worker that periodically checks the queue, takes a job, and processes it:

// openai/jobs/worker.ts
import { getJob, updateJob } from './queue';
import { emitJobEvent } from './events';

async function processJob(jobId: string) {
  const job = getJob(jobId);
  if (!job) return;

  updateJob(jobId, { status: 'in_progress' });
  await emitJobEvent(jobId, 'job.started', {});

  try {
    // Call the long-running business logic here
    const result = await doDeepGiftAnalysis(job.id, job.payload);

    updateJob(jobId, { status: 'completed', result });
    await emitJobEvent(jobId, 'job.completed', { resultSummary: summarize(result) });
  } catch (err) {
    updateJob(jobId, {
      status: 'failed',
      error: (err as Error).message,
    });
    await emitJobEvent(jobId, 'job.failed', { error: 'Internal error' });
  }
}

And the “looping” worker you can start somewhere at app startup:

// openai/jobs/workerLoop.ts
import { queue } from './queue';
import { processJob } from './worker';

export function startWorkerLoop() {
  setInterval(async () => {
    const jobId = queue.shift(); // strictly speaking, needs race-condition protection
    if (!jobId) return;

    await processJob(jobId);
  }, 1000); // check the queue once a second
}

This is a training example. In the real world, instead of setInterval you’d use a proper queue that “wakes” the worker when a new message arrives. But the core idea is clear: the worker is decoupled from the MCP tool, runs in the background, and talks to the MCP server via events.

6. Emitting MCP events from the worker

In previous lectures you’ve already seen the MCP event format: type, unique event_id, timestamp, job_id, and payload. Now we’ll show how a worker can call the helper emitJobEvent, which then delivers events to ChatGPT via the MCP server’s SSE channel.

Example of a simple helper:

// openai/jobs/events.ts
import { randomUUID } from 'crypto';
import { sendMcpEvent } from '../mcp/eventBus';

export async function emitJobEvent(
  jobId: string,
  type: 'job.started' | 'job.progress' | 'job.completed' | 'job.failed',
  payload: unknown
) {
  const event = {
    event_id: randomUUID(),
    type,
    job_id: jobId,
    timestamp: new Date().toISOString(),
    payload,
  };

  await sendMcpEvent(event);
}

And inside the MCP server, sendMcpEvent already knows how to push this event into SSEServerTransport from the MCP SDK: for example, via a local event bus or Redis Pub/Sub, as we covered in module 12.

Key idea: the worker doesn’t talk to ChatGPT directly. It talks to the MCP server, which holds the SSE connections and forwards events to clients.

7. Progress and partial results from the worker

Now the fun part: progress and partial results. In GiftGenius, the long analysis can be broken into stages:

• data collection and normalization;
• building basic segments;
• generating initial gift ideas;
• final ranking and textual explanation.

At each stage we can send job.progress and sometimes job.partial so the UI can already display the first gifts.

A notional worker:

async function doDeepGiftAnalysis(jobId: string, payload: GiftJob['payload']) {
  await emitJobEvent(jobId, 'job.progress', { step: 1, totalSteps: 4 });

  const normalized = await collectAndNormalizeData(payload);
  await emitJobEvent(jobId, 'job.progress', { step: 2, totalSteps: 4 });

  const roughGifts = await generateInitialGifts(normalized);
  await emitJobEvent(jobId, 'job.partial', { gifts: roughGifts.slice(0, 3) });

  await emitJobEvent(jobId, 'job.progress', { step: 3, totalSteps: 4 });

  const finalGifts = await rerankAndBeautify(roughGifts);
  await emitJobEvent(jobId, 'job.progress', { step: 4, totalSteps: 4 });

  return finalGifts;
}

Listening to events, the widget can first show 3 “draft” gifts with a note “Refining…,” and after job.completed — update the list and remove the loading indicator. This fits perfectly with the UX patterns we discussed in lecture 3.

8. Retry logic for workers

Now the most nerve‑wracking part: errors and retries.

Imagine the worker reaches out to an external product catalog API during processing, and it periodically responds with 500 or 429. Dropping the task after the first error is odd. But retrying forever is also unacceptable — you’ll DDoS yourself or the third‑party service.

We need a retry strategy with exponential backoff and a limit on the number of attempts.

Let’s start with an error classification that we’ll use later in the course:

• transient — timeouts, 500, 503, 429;
• permanent — invalid input, non‑existent resource;
• bug — code bugs, TypeError, unexpected exceptions.

Only transient errors should be retried. The rest should be honestly marked as 'failed'.

We’ll simplify and create a helper:

// openai/jobs/retry.ts
export function shouldRetry(error: unknown): boolean {
  if (!(error instanceof Error)) return false;
  // Roughly: HTTP 5xx or 429
  return /5\d\d|429/.test(error.message);
}

export function getDelayMs(base: number, attempt: number): number {
  const jitter = Math.random() * 100;   // small jitter
  return base * 2 ** attempt + jitter; // exponential backoff
}

Now let’s update the worker to account for attempts in GiftJob:

// openai/jobs/worker.ts
import { getJob, updateJob } from './queue';
import { emitJobEvent } from './events';
import { shouldRetry, getDelayMs } from './retry';

const MAX_ATTEMPTS = 5;

export async function processJob(jobId: string) {
  const job = getJob(jobId);
  if (!job) return;

  updateJob(jobId, { status: 'in_progress' });

  try {
    const result = await doDeepGiftAnalysis(job.id, job.payload);

    updateJob(jobId, { status: 'completed', result });
    await emitJobEvent(jobId, 'job.completed', {
      resultSummary: summarize(result),
    });
  } catch (err) {
    const attempts = job.attempts + 1;
    const error = err as Error;

    if (attempts <= MAX_ATTEMPTS && shouldRetry(error)) {
      const delay = getDelayMs(1000, attempts); // 1s,2s,4s...

      updateJob(jobId, { attempts, status: 'pending', error: error.message });

      setTimeout(() => {
        // In a real queue you'd re-enqueue the job with a delay
        processJob(jobId);
      }, delay);

      await emitJobEvent(jobId, 'job.progress', {
        retry: attempts,
        nextAttemptInMs: delay,
      });
    } else {
      updateJob(jobId, { status: 'failed', error: error.message });
      await emitJobEvent(jobId, 'job.failed', {
        error: 'Could not complete the analysis after several attempts',
      });
    }
  }
}

Several important points here.

First, attempts is stored in the job itself — convenient for logging and observability (you’ll clearly see on a graph how many jobs go through with retries).

Second, on each retry we send job.progress explicitly indicating that this is attempt #N. The model can use this information to tell the user that “the gift server is unstable; trying again.”

Third, we guarantee that in any case either job.completed or job.failed will be sent. No zombie jobs “neither alive nor dead.”

Cancellation ('canceled') is another important status. We won’t implement it in the training examples, but in production it’s usually set by the user (an “Cancel” button in the widget) or by a timeout. In that case, the worker, upon next taking the job from the queue, sees status: 'canceled', doesn’t start processing, and the MCP server sends a final job.canceled event.

9. Idempotency and retry: don’t step on the same rake twice

When you introduce retries, you immediately risk “doing the same thing twice.” In commerce modules this is critical (for example, double-charging), and even in GiftGenius there are scenarios where duplication is bad: sending two identical emails to a friend, duplicating a record in your internal analytics, etc.

So aim for two principles.

First: the job handler must be idempotent.

If you call it with the same jobId multiple times (within retries or by mistake), the world shouldn’t break. To achieve this:

• all side effects (DB writes, sending emails, creating orders) should be keyed by jobId or another natural identifier, so the code can quickly check whether we’ve already performed that step;
• if job.status is already 'completed' or 'failed', a repeated call can be ignored or simply return the already-prepared result.

Example of a simple guard:

export async function processJob(jobId: string) {
  const job = getJob(jobId);
  if (!job) return;

  if (job.status === 'completed' || job.status === 'failed') {
    // The job has already finished successfully or definitively failed
    return;
  }

  // ... remaining code
}

Second: events should also be idempotent.

We already discussed event_id and that the client can filter duplicates, but you should also be careful server-side: on worker restarts or queue recovery, don’t spam the client with identical job.progress events unnecessarily.

10. Where queues and workers live in your architecture

The diagram looks neat, but where does the worker physically run? There are several common options.

Integrated worker: the MCP server and worker are the same process/deploy. It accepts tool calls and runs the worker loop. The plus is simplicity: fewer services, easier deployment. The minus is scaling: to add workers, you have to scale the entire MCP server.

Dedicated worker: the MCP server is one service, workers are another. Between them is a queue and possibly Pub/Sub for events. This is often discussed in the context of BullMQ/Redis and MCP events: the MCP server subscribes to a Redis channel 'mcp:events', workers publish events there.

Hybrid: one instance of the MCP server also runs a worker; the other instances handle only HTTP/SSE. This can be useful if you deploy on Vercel or another serverless platform where always-on background processes are less than ideal.

In our training GiftGenius, the first option is acceptable for now: MCP server + one simple worker in-process. When you get to the modules on production and scaling, you can migrate workers to a separate service.

11. Example: full GiftGenius async pipeline

Let’s walk through what happens when a user writes in chat:

“I need a sophisticated gift selection for a space fan, taking their past purchases into account.”

1. The model decides to call the start_deep_analysis tool with the recipient’s profile and budget parameters.
2. The tool creates a GiftJob in the DB with status 'pending', enqueues it, and returns the jobId + a confirmation message.
3. ChatGPT explains to the user that the analysis has started and can pass the jobId to the GiftGenius widget.
4. The widget subscribes to events for this jobId via SSE, shows a progress bar and a status like “Collecting and analyzing data.”
5. The worker, seeing a new job in the queue, updates the status to 'in_progress' and sends job.started.
6. During processing it sends several job.progress events (stages) and job.partial (the first 2–3 gifts).
7. If an external API fails along the way, the worker retries with exponential backoff, updating attempts and sending an event with retry information.
8. At the end it either sends job.completed with a short summary and final recommendations, or job.failed with a clear explanation.
9. The widget updates the UI based on these events, and ChatGPT can form a textual summary and suggest a follow‑up: “Show more ideas,” “Narrow the budget,” “Change the gift type.”

From the user’s perspective, this is a “live” long-running process under control. From the backend’s perspective, it’s a normal async pipeline with a queue, workers, and retries.

12. A small exercise (for self‑practice)

If you want to reinforce the material, try the following for GiftGenius:

• design a jobs table schema for a real DB: which indexes you need, which fields will be used in filtering (by user, by status, by creation date);
• sketch a TypeScript type for an HTTP endpoint /api/jobs/:id so the widget can, in the worst case, poll status if SSE is unavailable;
• describe a retry policy: number of attempts, base delay, what to do with tasks that still fail (a simple dead‑letter table or logging + alert).

You’ll find this useful later when, in the production and observability modules, we talk about metrics like “how many jobs have been stuck in pending longer than N minutes.”

13. Common mistakes when working with asynchronous jobs

Mistake #1: doing everything synchronously in a tool call.
The most common trap is trying to cram all the heavy work into a single MCP tool without a queue. While the request volume is small, this seems to work. As soon as load grows or external APIs slow down, you get timeouts, a stalling chat, and a very anxious UX. Any operation that can potentially last tens of seconds or more is better designed from the start as an async job with a jobId.

Mistake #2: no explicit Job model.
Sometimes developers try to get by with “just messages in a queue,” without storing task state in a DB. As a result, it’s hard to answer basic questions: “what’s the task’s status?”, “how many times have we tried to execute it?”, “why did it fail?”. A clear Job model with fields like status, attempts, error, createdAt is the foundation for debugging, monitoring, and UX.

Mistake #3: no retries or, conversely, infinite retries.
Some skip retries entirely and fail on the first 500; others write while (!success) and don’t cap attempts. In the first case, you lose many tasks due to brief outages; in the second, you create load “storms” and risk blocking external APIs. You need a sensible middle ground: a limited number of attempts + exponential backoff + separating transient and permanent errors.

Mistake #4: non‑idempotent handlers.
If on every attempt you, for example, create a new record in a third‑party system without checking, execute the same payment, or send the same email — retries quickly become a problem. The handler should be able to tell that the task with this jobId was already successfully completed and must not repeat dangerous side effects.

Mistake #5: no events on errors.
Sometimes a worker crashes with an unexpected exception, logs it to the console, and that’s it. The user sits and waits forever for job.completed, not knowing everything died long ago. Any branch where processing ended with an error must ultimately result in job.failed and an updated Job status in the DB. Without this, your MCP streams turn into a one‑way “black box.”

Mistake #6: progress events that are too frequent.
The desire to “be honest” and send job.progress for every one percent of completion overloads the network, the client, and the MCP server. It’s better to report progress on stage changes or on a large delta (for example, every 10%); keep everything else only in internal logs.

Mistake #7: using an in‑memory queue in production.
The training example with queue: string[] and Map is good for understanding the architecture, but in a real production system it will fall apart on the first process restart or server crash. For serious operation you need external queues and storage: SQS, Pub/Sub, RabbitMQ, Redis Streams, etc. In‑memory variants are suitable only for local development and simple demos.