Tareas asíncronas: colas, workers y reintentos (retry)

1. ¿Por qué tareas asíncronas en ChatGPT App?

Si el mundo fuera ideal, cualquier herramienta MCP terminaría en un par de cientos de milisegundos. Pero en la vida, todo lo interesante —es largo y pesado—:

• procesamiento de un CSV grande con el historial de compras del usuario;
• agregación de datos de varias API externas, cada una de las cuales a veces «duerme» y a veces responde con 503;
• construcción de recomendaciones complejas con muchos pasos intermedios;
• generación de informes y presentaciones voluminosos.

Si intentas meter todo esto en una sola llamada síncrona del tool, te toparás con tres problemas.

Primero, timeouts. La sesión de ChatGPT, la infraestructura HTTP y el cliente MCP —nada de esto está pensado para respuestas «dentro de cinco minutos». Un servidor que mantiene la conexión demasiado tiempo parecerá «colgado» tanto para ChatGPT como para el usuario.

Segundo, gestión de carga. Si cien usuarios lanzan a la vez el «súper análisis de regalos de Navidad», no quieres que el servidor MCP mantenga cien tareas largas de forma síncrona en hilos HTTP. Necesitas una capa que pueda absorber picos, distribuir tareas en una cola y procesarlas con varios workers.

Tercero, UX. El usuario pulsa un botón en el widget GiftGenius y se queda mirando un solo spinner durante 40 segundos: la sensación recuerda a los viejos bancos online. Es mucho más agradable el modelo «respuesta rápida + progreso + posibilidad de cancelar».

Estos problemas se resuelven con el esquema: «inicio → cola → background → eventos».

2. Arquitectura básica de async‑job en el contexto de MCP

Tomemos nuestro GiftGenius. Supongamos que aparece un nuevo escenario pesado: «Análisis profundo de preferencias a partir del historial de compras y redes sociales de un amigo». Algo así puede tardar varios minutos, por lo que:

1. La herramienta MCP (tool) recibe los parámetros de la solicitud desde el modelo.
2. En lugar de calcularlo todo de inmediato, crea un registro Job en la base de datos.
3. Encola la tarea.
4. Responde inmediatamente a ChatGPT: «Análisis iniciado, aquí tienes el jobId».
5. Un worker en segundo plano saca la tarea de la cola, realiza el trabajo pesado y, a medida que avanza, envía eventos MCP job.progress y job.partial, y al final — job.completed o job.failed.

Desde el punto de vista de arquitectura, se ve así:

flowchart LR
    subgraph ChatGPT
      U[Usuario] --> GPT[Modelo + UI de ChatGPT]
    end

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

    subgraph Backend
      MCP -->|crear Job| DB[(BD de tareas)]
      MCP -->|enqueue| Q[Cola]
      W[Worker] -->|take job| Q
      W -->|update status/progress| DB
      W -->|MCP events: job.progress/job.completed| MCP
    end

    MCP -->|SSE events| GPT

Idea clave: el servidor MCP no tiene por qué ser un monolito. A menudo actúa como un «fachada» sobre tu infraestructura asíncrona interna: recibe tool‑calls, crea jobs y envía eventos, mientras que el trabajo pesado lo realizan procesos workers separados.

3. Modelo de datos para una tarea asíncrona

Empecemos con un modelo simple Job. Usaremos TypeScript y un servidor Node/MCP hipotético para que veas cómo encaja en tu stack.

Un modelo sencillo en memoria/BD puede verse así:

// 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;  // texto/ID del perfil
    budget: number;
  };
  result?: unknown;          // recomendaciones finales
  error?: string;            // motivo del error
  attempts: number;          // cuántas veces se intentó ejecutar
  createdAt: Date;
  updatedAt: Date;
}

En un proyecto real guardarás GiftJob en Postgres, DynamoDB, Firestore o similar, pero para la lección nos importan los campos:

• status — estado actual de la tarea, que aparecerá tanto en los eventos como en el UX;
• attempts — contador para los reintentos;
• error — para logs y depuración;
• payload — datos de entrada que el worker usa para el procesamiento.

4. La herramienta MCP que crea el async‑job

Imaginemos la herramienta start_deep_analysis. Antes podría hacerlo todo de forma síncrona; ahora solo pone la tarea en la cola y devuelve el jobId.

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

// Tipos ficticios para el 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: `He iniciado un análisis profundo. ID de la tarea: ${jobId}. Enviaré actualizaciones a medida que estén listas.`,
  };
}

Importante:

• La herramienta MCP trabaja rápido: como mucho un par de operaciones a la BD/cola;
• devuelve una respuesta estructurada con jobId, que ChatGPT puede usar en su «explicación al usuario» y que el widget GiftGenius puede guardar en su widgetState.

Tu JSON Schema para esta herramienta simplemente describe jobId como cadena y message como texto legible — el modelo entenderá que es un identificador de tarea y podrá referenciarlo en los siguientes pasos del diálogo.

5. Cola y worker simples: versión didáctica

Para no traer ahora Redis, RabbitMQ y demás, haremos una cola en memoria simplificada. En producción, por supuesto, será un servicio aparte (SQS/BullMQ/Cloud Tasks, etc.), pero la lógica será la misma.

Primero, un esqueleto de cola:

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

const jobs = new Map<string, GiftJob>();   // "BD" en memoria
export const queue: string[] = [];         // cola simplificada por 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);
}

Ahora un worker primitivo que mira periódicamente la cola, toma un job y lo procesa:

// 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 {
    // Aquí llamamos a la lógica de negocio larga
    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' });
  }
}

Y el worker «cíclico» que puede arrancarse al iniciar la aplicación:

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

export function startWorkerLoop() {
  setInterval(async () => {
    const jobId = queue.shift(); // en rigor haría falta protección contra condiciones de carrera
    if (!jobId) return;

    await processJob(jobId);
  }, 1000); // comprobamos la cola una vez por segundo
}

Es un ejemplo didáctico. En el mundo real, en lugar de setInterval tendrás una cola seria que «despierta» al worker cuando aparece un nuevo mensaje. Pero la idea general se ve clara: el worker está separado de la herramienta MCP, funciona en background y habla con el servidor MCP mediante eventos.

6. Generación de eventos MCP desde el worker

En lecciones anteriores ya viste el formato de los eventos MCP: tipo (type), event_id único, timestamp, job_id y payload. Ahora muestra­remos cómo el worker puede llamar al helper emitJobEvent, y este entregar los eventos a ChatGPT a través del canal SSE del servidor MCP.

Ejemplo de helper sencillo:

// 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);
}

Y sendMcpEvent dentro del servidor MCP ya sabe cómo inyectar este evento en SSEServerTransport del MCP SDK: por ejemplo, mediante un bus de eventos local o Redis Pub/Sub, como vimos en el módulo 12.

Idea clave: el worker no se comunica con ChatGPT directamente. Se comunica con el servidor MCP, y este mantiene las conexiones SSE y reenvía los eventos a los clientes.

7. Progreso y partial results desde el worker

Ahora lo más interesante: progreso y resultados parciales. En GiftGenius, un análisis largo puede dividirse en etapas:

• recopilación y normalización de datos;
• construcción de segmentos básicos;
• generación de ideas iniciales de regalos;
• ranking final y explicación en texto.

En cada etapa podemos enviar job.progress y, a veces, job.partial, para que el UI muestre ya los primeros regalos.

Un worker hipotético:

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;
}

El widget, al escuchar los eventos, puede mostrar primero 3 regalos «borrador» con la etiqueta «Afinando aún», y después de job.completed — actualizar la lista y quitar el indicador de carga. Todo esto encaja perfectamente con los patrones de UX de los que hablamos en la lección 3.

8. Lógica de retry para los workers

Ahora la parte más delicada: errores y reintentos.

Imagina que el worker, al procesar la tarea, llama a una API externa de catálogo de productos y esta responde periódicamente con 500 o 429. Abandonar la tarea tras el primer error es extraño. Pero reintentar indefinidamente tampoco vale: acabarías haciendo DDoS a tu propio sistema o al servicio externo.

Necesitamos una estrategia de retry con retardo exponencial y límite de intentos.

Empecemos por una clasificación de errores, que nos servirá también más adelante en el curso:

• temporales (transient) — timeouts, 500, 503, 429;
• permanentes (permanent) — entrada no válida, recurso inexistente;
• fatales (bug) — bugs de código, TypeError, excepción inesperada.

Solo tiene sentido repetir ante errores temporales. El resto hay que marcarlos honestamente como 'failed'.

Simplifiquemos y creemos un helper:

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

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

Ahora actualizamos el worker para que tenga en cuenta attempts en 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(() => {
        // En una cola real volverías a reencolar el job con retraso
        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: 'No se pudo completar el análisis después de varios intentos',
      });
    }
  }
}

Hay varios puntos importantes.

Primero, attempts se guarda en la propia tarea —es útil tanto para el logging como para la observabilidad (en una gráfica se verá claramente cuántas tareas pasan con reintentos).

Segundo, en cada retry enviamos job.progress indicando explícitamente que es el intento n.º N. El modelo puede usar esta información para explicar al usuario que «el servidor de regalos responde de forma inestable; voy a intentarlo de nuevo».

Tercero, garantizamos que, en cualquier caso, se enviará o bien job.completed o bien job.failed. Nada de tareas colgando «ni vivas ni muertas».

La cancelación ('canceled') es otro estado importante. En los ejemplos didácticos no la implementamos, pero en producción suele iniciarse por el usuario (botón «Cancelar» en el widget) o por timeout. En ese caso, el worker, al tomar la siguiente vez la tarea de la cola, ve status: 'canceled', no inicia el procesamiento y el servidor MCP envía el evento final job.canceled.

9. Idempotencia y retry: no tropezar dos veces con la misma piedra

Cuando introduces retry, aparece el riesgo de «hacer lo mismo dos veces». En módulos de comercio es crítico (por ejemplo, un doble cargo), y también en GiftGenius hay casos en los que repetir es malo: enviar dos correos iguales al amigo, duplicar una entrada en tu analítica interna, etc.

Por eso, hay que seguir dos principios.

Primero: el handler del job debe ser idempotente.

Si lo llamas con el mismo jobId varias veces (en el marco de un retry o por error), el mundo no debe romperse. Para lograrlo:

• todos los efectos secundarios (escritura en la BD, envío de correos, creación de pedidos) deben estar ligados al jobId u otro identificador natural, de forma que el código pueda comprobar rápidamente si ya hicimos ese paso;
• si job.status ya es 'completed' o 'failed', puedes ignorar la llamada repetida o devolver simplemente el resultado ya preparado.

Ejemplo de protección simple:

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

  if (job.status === 'completed' || job.status === 'failed') {
    // La tarea ya se completó correctamente o terminó definitivamente
    return;
  }

  // ... resto del código
}

Segundo: los eventos también deben ser idempotentes.

Ya hablamos de event_id y de que el cliente puede filtrar duplicados, pero del lado del servidor también hay que ser cuidadoso: al reiniciar el worker o al recuperar desde la cola, no satures al cliente con los mismos job.progress sin necesidad.

10. ¿Dónde están las colas y los workers en tu arquitectura?

En el diagrama todo es bonito, pero ¿dónde se ejecuta físicamente el worker? Hay varias opciones típicas.

Worker integrado: servidor MCP y worker —el mismo proceso/despliegue. Recibe los tool‑calls y levanta el worker loop. Ventaja: simplicidad; menos servicios, despliegue más fácil. Desventaja: escalado; para añadir workers, tienes que escalar todo el servidor MCP.

Worker dedicado: el servidor MCP es un servicio, los workers otro. Entre ambos — una cola y, posiblemente, Pub/Sub para eventos. Es lo que se comenta mucho en el contexto de BullMQ/Redis y eventos MCP: el servidor MCP está suscrito al canal de Redis 'mcp:events', y los workers publican ahí los eventos.

Variante combinada: una instancia del servidor MCP ejecuta también un worker, las demás instancias — solo HTTP/SSE. Puede ser útil si despliegas en Vercel u otra plataforma serverless, donde los procesos en background constantes son poco cómodos.

En nuestro GiftGenius didáctico de momento vale la primera opción: servidor MCP + un worker simple en el mismo proceso. Cuando llegues a los módulos sobre producción y escalado, podrás migrar los workers a un servicio separado.

11. Ejemplo: pipeline async completo de GiftGenius

Repasemos de forma conectada qué ocurre cuando el usuario en el chat escribe:

«Necesito una selección compleja de regalos para un fan del espacio, teniendo en cuenta sus compras anteriores».

1. El modelo decide llamar a la herramienta start_deep_analysis con los parámetros del perfil del destinatario y el presupuesto.
2. La herramienta crea un GiftJob en la BD con estado 'pending', lo pone en la cola y devuelve el jobId + un mensaje de confirmación.
3. ChatGPT explica al usuario que el análisis se ha iniciado y puede pasar el jobId al widget GiftGenius.
4. El widget se suscribe a los eventos de ese jobId vía SSE, muestra una barra de progreso y el estado «Recopilando y analizando datos».
5. El worker, al ver un nuevo job en la cola, actualiza el estado a 'in_progress' y envía job.started.
6. Durante el trabajo, envía varias veces job.progress (etapas) y job.partial (los primeros 2–3 regalos).
7. Si por el camino falla una API externa, el worker lo intenta de nuevo con backoff exponencial, actualizando attempts y enviando un evento con información sobre el reintento.
8. Al final, o bien envía job.completed con un resumen breve y recomendaciones finales, o job.failed con una explicación clara.
9. El widget, basándose en esos eventos, actualiza el UI, y ChatGPT puede generar un resumen y proponer un follow‑up: «Mostrar más ideas», «Reducir el presupuesto», «Cambiar el tipo de regalo».

Desde el punto de vista del usuario, es un proceso largo «vivo» y bajo control. Desde el punto de vista del backend — un pipeline asíncrono normal con cola, workers y retry.

12. Pequeño ejercicio (para práctica autónoma)

Si quieres afianzar el material, intenta para GiftGenius:

• idear el esquema de la tabla jobs para una BD real: qué índices necesitas, qué campos participarán en el filtrado (por usuario, por estado, por fecha de creación);
• esbozar el tipo TypeScript para el endpoint HTTP /api/jobs/:id, para que el widget pueda, en el peor de los casos, hacer polling del estado si SSE no está disponible;
• describir la política de retry: cuántos intentos, retardo base, qué hacer con las tareas que igualmente fallan (tabla dead‑letter simple o logging + alerta).

Este ejercicio te resultará útil más adelante, cuando en los módulos sobre producción y observabilidad hablemos de métricas como «cuántas tareas han quedado en estado pending más de N minutos».

13. Errores típicos al trabajar con tareas asíncronas

Error n.º 1: hacerlo todo de forma síncrona en el tool‑call.
La trampa más común es intentar meter todo el trabajo pesado en una sola herramienta MCP sin cola. Mientras haya pocas solicitudes, parece que funciona. En cuanto la carga crece o las API externas empiezan a ir lentas, sufres timeouts, bloqueos del chat y un UX muy nervioso. Cualquier operación que potencialmente pueda durar decenas de segundos o más es mejor diseñarla desde el inicio como async‑job con jobId.

Error n.º 2: ausencia de un modelo Job explícito.
A veces los desarrolladores intentan apañarse «solo con mensajes en la cola», sin guardar el estado de las tareas en la BD. Al final es difícil responder a preguntas básicas: «¿cuál es el estado de la tarea?», «¿cuántas veces intentamos ejecutarla?», «¿por qué falló?». Un modelo claro Job con campos status, attempts, error, createdAt — es la base para depuración, monitorización y UX.

Error n.º 3: ausencia de retry o, al contrario, reintentos infinitos.
Algunos no hacen retry y caen al primer 500, otros hacen while (!success) y no limitan el número de intentos. En el primer caso pierdes muchas tareas por fallos breves; en el segundo, generas «tormentas» de carga y corres el riesgo de bloquear las API externas. Necesitas un término medio razonable: número limitado de intentos + retardo exponencial + separación de errores temporales y permanentes.

Error n.º 4: handlers no idempotentes.
Si en cada intento, por ejemplo, creas un nuevo registro en un sistema externo sin comprobar, ejecutas el mismo pago o envías el mismo correo — el retry se convierte rápidamente en un problema. El handler debe poder entender que la tarea con ese jobId ya se llevó a buen término y no repetir los efectos secundarios peligrosos.

Error n.º 5: ausencia de eventos en caso de error.
A veces el worker cae con una excepción inesperada, la registra en consola y ahí se queda. El usuario se queda esperando eternamente job.completed, sin saber que todo murió hace tiempo. Cualquier rama donde el procesamiento termine con error debe desembocar en job.failed y en la actualización del estado del Job en la BD. Sin eso, tus flujos MCP se convierten en una «caja negra» unidireccional.

Error n.º 6: eventos de progreso demasiado frecuentes.
La intención de «ser honesto» y enviar job.progress por cada uno por ciento de avance provoca sobrecarga de red, del cliente y del servidor MCP. Es mejor enviar el progreso en cambios de etapa o por deltas grandes (por ejemplo, cada 10%), y el resto guardarlo solo en logs internos.

Error n.º 7: usar una cola en memoria en producción.
El ejemplo didáctico con queue: string[] y Map es bueno para entender la arquitectura, pero en un sistema de producción real se caerá al primer reinicio del proceso o caída del servidor. Para explotación seria hacen falta colas y almacenes externos: SQS, Pub/Sub, RabbitMQ, Redis Streams, etc. Las variantes en memoria sirven solo para desarrollo local y demos sencillas.