Tarefas assíncronas: filas, workers, reenvios (retry)

1. Por que tarefas assíncronas no ChatGPT App

Se o mundo fosse ideal, qualquer MCP tool terminaria em algumas centenas de milissegundos. Mas, na vida real, tudo que é interessante é demorado e pesado:

• processar um CSV grande com o histórico de compras do usuário;
• agregar dados de várias APIs externas, cada uma ora “dorme”, ora responde com 503;
• construir recomendações complexas com um monte de etapas intermediárias;
• gerar relatórios e apresentações grandes.

Se você tentar enfiar tudo isso em uma única chamada síncrona (tool‑call), vai esbarrar em três problemas.

Primeiro, timeouts. A sessão do ChatGPT, a infraestrutura HTTP, o cliente MCP — nada disso é feito para responder “daqui a cinco minutos”. Um servidor que mantém a conexão tempo demais vai parecer “travado” tanto para o ChatGPT quanto para o usuário.

Segundo, controle de carga. Se cem usuários executarem ao mesmo tempo o “super‑analisador de presentes de fim de ano”, você não vai querer que o servidor MCP mantenha cem tarefas longas síncronas dentro das threads HTTP. Você precisa de uma camada que absorva picos, distribua as tarefas em uma fila e as processe com vários workers.

Terceiro, UX. O usuário clica no botão do widget GiftGenius e fica olhando para um único spinner por 40 segundos — a sensação é de internet banking antigo. É muito melhor o modelo “resposta rápida + progresso + opção de cancelar”.

Esses problemas se resolvem com o esquema geral: “disparo → fila → segundo plano → eventos”.

2. Arquitetura básica de async‑job no contexto de MCP

Vamos pegar o nosso GiftGenius. Suponha que surgiu um novo cenário pesado: “Análise profunda de preferências com base no histórico de compras e nas redes sociais do amigo”. Isso pode levar alguns minutos, então:

1. A ferramenta MCP (tool) recebe os parâmetros do pedido vindos do modelo.
2. Em vez de calcular tudo na hora, ela cria um registro Job no banco.
3. Coloca a tarefa na fila.
4. Responde imediatamente ao ChatGPT: “Análise iniciada, aqui está o jobId”.
5. Um worker em segundo plano pega a tarefa da fila, executa o trabalho pesado e, ao longo do caminho, envia eventos MCP job.progress e job.partial, no final — job.completed ou job.failed.

Do ponto de vista da arquitetura, isso se parece com o seguinte:

flowchart LR
    subgraph ChatGPT
      U[Usuário] --> GPT[Modelo + ChatGPT UI]
    end

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

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

    MCP -->|SSE events| GPT

Ideia importante: o servidor MCP não precisa ser um monólito. Muitas vezes ele atua como um façade sobre sua infraestrutura assíncrona interna: ele recebe tool‑calls, cria jobs e envia eventos, enquanto o trabalho pesado é feito por processos/workers separados.

3. Modelo de dados para uma tarefa assíncrona

Vamos começar com um modelo simples Job. Vamos usar TypeScript e um servidor Node/MCP hipotético, para você ver como isso se encaixa no seu stack.

Um modelo mínimo em memória/BD pode ser assim:

// 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 do perfil
    budget: number;
  };
  result?: unknown;          // recomendações finais
  error?: string;            // motivo do erro
  attempts: number;          // quantas vezes tentamos executar
  createdAt: Date;
  updatedAt: Date;
}

Em um projeto real, você vai armazenar GiftJob no Postgres, DynamoDB, Firestore ou outro, mas, para a aula, os campos importantes são:

• status — o estado atual da tarefa, que aparecerá tanto nos eventos quanto no UX;
• attempts — contador para retry;
• error — para logs e depuração;
• payload — dados de entrada que o worker usa no processamento.

4. Ferramenta MCP que cria um async‑job

Imagine a ferramenta start_deep_analysis. Antes ela poderia fazer tudo de forma síncrona; agora, apenas coloca a tarefa na fila e retorna o jobId.

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

// Pseudotipos para o 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: `Iniciei a análise profunda. ID da tarefa: ${jobId}. Vou enviar atualizações conforme forem ficando prontas.`,
  };
}

Aqui é importante que:

• A ferramenta MCP seja rápida: no máximo algumas chamadas ao banco/fila;
• ela retorne uma resposta estruturada com o jobId, que o ChatGPT pode usar na sua “explicação para o usuário” e que o widget GiftGenius pode salvar no seu widgetState.

O seu JSON Schema para essa ferramenta apenas descreve o jobId como string e message como texto legível — o modelo entenderá que é um identificador de tarefa e poderá referenciá‑lo nos próximos passos do diálogo.

5. Fila simples e worker: versão didática

Para não puxar agora Redis, RabbitMQ e afins, vamos fazer uma fila em memória simplificada. Em produção, claro, isso vira um serviço separado (SQS/BullMQ/Cloud Tasks etc.), mas a lógica é a mesma.

Primeiro, um esboço de fila:

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

const jobs = new Map<string, GiftJob>();   // "BD" em memória
export const queue: string[] = [];         // fila 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);
}

Agora um worker bem simples, que periodicamente olha a fila, pega uma job e a processa:

// 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 {
    // Aqui chamamos a lógica de negócio demorada
    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' });
  }
}

E o próprio “worker cíclico”, que pode ser iniciado em algum ponto do app:

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

export function startWorkerLoop() {
  setInterval(async () => {
    const jobId = queue.shift(); // o ideal seria ter proteção contra condições de corrida
    if (!jobId) return;

    await processJob(jobId);
  }, 1000); // verificamos a fila uma vez por segundo
}

Este é um exemplo didático. No mundo real, em vez de setInterval haverá uma fila “de verdade” que “acorda” o worker quando chega uma nova mensagem. Mas a ideia geral fica clara: o worker é separado da ferramenta MCP, roda em segundo plano e conversa com o servidor MCP via eventos.

6. Geração de eventos MCP a partir do worker

Nas aulas anteriores você já viu o formato de eventos MCP: tipo (type), event_id único, timestamp, job_id e payload. Agora mostramos como o worker pode chamar o helper emitJobEvent, e este entregar os eventos ao ChatGPT por meio do canal SSE do servidor MCP.

Exemplo de um helper simples:

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

E o sendMcpEvent dentro do servidor MCP já sabe como empurrar esse evento para o SSEServerTransport do MCP SDK: por exemplo, via um barramento local de eventos ou Redis Pub/Sub, como vimos no módulo 12.

Ponto-chave: o worker não fala diretamente com o ChatGPT. Ele fala com o servidor MCP, que mantém as conexões SSE e retransmite os eventos aos clientes.

7. Progresso e resultados parciais vindos do worker

Agora o mais saboroso: progresso e resultados parciais. No GiftGenius, a análise longa pode ser dividida em etapas:

• coleta e normalização de dados;
• construção de segmentos básicos;
• geração de ideias iniciais de presentes;
• ranqueamento final e explicação em texto.

Em cada etapa podemos enviar job.progress e às vezes — job.partial, para que o UI já mostre os primeiros presentes.

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

O widget, ouvindo os eventos, pode primeiro mostrar 3 presentes “rascunho” com a etiqueta “Ainda refinando”, e depois do job.completed — atualizar a lista e remover o indicador de carregamento. Tudo isso se encaixa perfeitamente nos padrões de UX de que falamos na aula 3.

8. Lógica de retry para workers

Agora a parte mais tensa: erros e repetições.

Imagine que, ao processar a tarefa, o worker chama uma API externa de catálogo de produtos e ela, de tempos em tempos, responde com 500 ou 429. Derrubar a tarefa no primeiro erro é estranho. Mas também não dá para repetir infinitamente: você causa DDoS em si mesmo ou no serviço de terceiros.

Precisamos de uma estratégia de retry com atraso exponencial e limitação de tentativas.

Comecemos pela classificação de erros, que será útil no restante do curso:

• temporários (transient) — timeouts, 500, 503, 429;
• permanentes (permanent) — entrada inválida, recurso inexistente;
• fatais (bug) — bugs de código, TypeError, exceção inesperada.

Só faz sentido repetir nos erros temporários. O restante deve ser marcado honestamente como 'failed'.

Vamos simplificar e criar um helper:

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

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

Agora vamos atualizar o worker para considerar attempts em 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(() => {
        // Em uma fila real você re-enfileiraria a job com atraso
        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: 'Não foi possível concluir a análise após várias tentativas',
      });
    }
  }
}

Alguns pontos importantes.

Primeiro, attempts fica armazenado na própria tarefa — é útil tanto para logging quanto para observabilidade (no gráfico, dá para ver claramente quantas tarefas passam com retries).

Segundo, a cada retry enviamos job.progress indicando explicitamente que é a tentativa nº N. O modelo pode usar essa informação para explicar ao usuário que “o servidor de presentes está instável; tentando novamente”.

Terceiro, garantimos que, em qualquer caso, será enviado job.completed ou job.failed. Nada de tarefas “presas no limbo”.

Cancelamento ('canceled') — é outro status importante. Nos exemplos didáticos não vamos implementá‑lo, mas, em produção, geralmente é acionado pelo usuário (botão “Cancelar” no widget) ou por timeout. Nesse caso, ao pegar a tarefa novamente, o worker vê status: 'canceled', não inicia o processamento, e o servidor MCP envia o evento final job.canceled.

9. Idempotência e retry: não tropece duas vezes na mesma pedra

Quando você introduz retry, surge o risco de “fazer a mesma coisa duas vezes”. Em módulos de commerce isso é crítico (por exemplo, cobrar em dobro), e no GiftGenius também há cenários em que repetir é ruim: enviar dois e‑mails iguais ao amigo, duplicar um registro na sua análise interna etc.

Portanto, siga dois princípios.

Primeiro: o handler da job deve ser idempotente.

Se você o chama com o mesmo jobId várias vezes (no contexto de retry ou por engano), o mundo não pode quebrar. Para isso:

• todos os efeitos colaterais (escrita no banco, envio de e‑mails, criação de pedidos) devem estar atrelados ao jobId ou outro identificador natural, para que o código possa verificar rapidamente se aquele passo já foi feito;
• se o job.status já for 'completed' ou 'failed', a chamada repetida pode ser ignorada ou apenas retornar o resultado já pronto.

Exemplo de uma proteção simples:

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

  if (job.status === 'completed' || job.status === 'failed') {
    // A job já foi concluída com sucesso ou encerrada definitivamente
    return;
  }

  // ... restante do código
}

Segundo: os próprios eventos também devem ser idempotentes.

Já falamos sobre event_id e que o cliente pode filtrar duplicados, mas no servidor também é bom ter cuidado: ao reiniciar o worker ou ao restaurar da fila, não faça spam para o cliente com os mesmos job.progress sem necessidade.

10. Onde ficam as filas e os workers na sua arquitetura

Na figura fica bonito, mas onde o worker roda fisicamente? Há algumas opções típicas.

Worker integrado: servidor MCP e worker — o mesmo processo/deploy. Ele recebe as tool‑calls e também inicia o worker loop. Vantagem: simplicidade; menos serviços, deploy mais fácil. Desvantagem — escalabilidade: para adicionar workers, você precisa escalar todo o servidor MCP.

Worker dedicado: servidor MCP — um serviço, workers — outro. Entre eles — uma fila e, possivelmente, Pub/Sub para eventos. É o que muito se vê com BullMQ/Redis e eventos MCP: o servidor MCP fica inscrito no canal Redis 'mcp:events', e os workers publicam eventos lá.

Variante combinada: uma instância do servidor MCP roda também um worker; as demais instâncias — apenas HTTP/SSE. Isso é útil se você faz deploy no Vercel ou outra plataforma serverless, onde processos contínuos em segundo plano não são a melhor experiência.

No nosso GiftGenius didático, por enquanto vale a primeira opção: servidor MCP + um worker simples no mesmo processo. Quando você chegar aos módulos de produção e escalabilidade, poderá migrar os workers para um serviço separado.

11. Exemplo: pipeline async completo do GiftGenius

Vamos percorrer, de ponta a ponta, o que acontece quando o usuário escreve no chat:

“Preciso de uma curadoria complexa de presentes para um fã de espaço, levando em conta suas compras anteriores”.

1. O modelo decide chamar a ferramenta start_deep_analysis com os parâmetros do perfil do destinatário e do orçamento.
2. A ferramenta cria um GiftJob no banco com status 'pending', coloca na fila e retorna jobId + uma mensagem de confirmação.
3. O ChatGPT explica ao usuário que a análise foi iniciada e pode passar o jobId ao widget do GiftGenius.
4. O widget assina os eventos desse jobId via SSE, mostra uma barra de progresso e o status “Coletando e analisando dados”.
5. Ao ver uma nova job na fila, o worker atualiza o status para 'in_progress' e envia job.started.
6. No processo, ele envia várias vezes job.progress (etapas) e job.partial (os primeiros 2–3 presentes).
7. Se alguma API externa cair no caminho, o worker tenta novamente com backoff exponencial, atualizando attempts e enviando um evento com informações sobre a nova tentativa.
8. No fim, ele envia job.completed com um resumo curto e as recomendações finais, ou job.failed com uma explicação clara.
9. Com base nesses eventos, o widget atualiza o UI, e o ChatGPT pode formar um sumário em texto e sugerir um follow‑up: “Mostrar mais ideias”, “Apertar o orçamento”, “Mudar o tipo de presente”.

Para o usuário, é um processo longo “vivo”, sob controle. Para o backend, é um pipeline assíncrono normal com fila, workers e retry.

12. Pequeno exercício (para prática individual)

Se quiser fixar o conteúdo, tente para o GiftGenius:

• imaginar o esquema da tabela jobs para um banco real: quais índices são necessários, quais campos participarão da filtragem (por usuário, por status, por data de criação);
• esboçar o tipo TypeScript do endpoint HTTP /api/jobs/:id, para que o widget possa, em último caso, fazer polling do status, se o SSE não estiver disponível;
• descrever a política de retry: quantas tentativas, atraso base, o que fazer com as tarefas que ainda assim falharam (uma tabela dead‑letter simples ou logging + alerta).

Este exercício será útil depois, quando, nos módulos sobre produção e observabilidade, falarmos sobre métricas como “quantas tarefas ficaram no status pending por mais de N minutos”.

13. Erros típicos ao trabalhar com tarefas assíncronas

Erro nº 1: fazer tudo de forma síncrona dentro do tool‑call.
A armadilha mais comum é tentar enfiar todo o trabalho pesado em uma única ferramenta MCP, sem fila. Enquanto há poucas requisições, até funciona. Assim que a carga cresce ou as APIs externas começam a atrasar, você pega timeouts, o chat fica travado e o UX fica bem nervoso. Qualquer operação que possa durar dezenas de segundos ou mais deve ser projetada desde o início como um async‑job com jobId.

Erro nº 2: ausência de um modelo Job explícito.
Às vezes, devs tentam se virar “só com mensagens na fila”, sem guardar o estado das tarefas no banco. Aí fica difícil responder perguntas básicas: “qual é o status da tarefa?”, “quantas vezes tentamos executá‑la?”, “por que ela caiu?”. Um modelo claro de Job com campos status, attempts, error, createdAt — é a base para debug, monitoração e UX.

Erro nº 3: sem retry ou, ao contrário, com repetições infinitas.
Alguns não fazem retry e caem no primeiro 500; outros fazem while (!success) e não limitam o número de tentativas. No primeiro caso, você perde um monte de tarefas por falhas temporárias; no segundo, cria “tempestades” de carga e corre o risco de bloquear APIs externas. É preciso um meio‑termo: número limitado de tentativas + atraso exponencial + separação entre erros temporários e permanentes.

Erro nº 4: handlers não idempotentes.
Se a cada tentativa você, por exemplo, cria um novo registro em um sistema externo sem checar, executa o mesmo pagamento ou envia o mesmo e‑mail — o retry rapidamente vira um problema. O handler deve conseguir identificar que a tarefa com esse jobId já foi concluída e não repetir efeitos colaterais perigosos.

Erro nº 5: ausência de eventos em caso de erro.
Acontece de o worker cair com uma exceção inesperada, logar no console e só. O usuário fica esperando para sempre o job.completed, sem saber que já morreu faz tempo. Qualquer ramificação que termine com erro deve, no fim, resultar em job.failed e atualização do status do Job no banco. Sem isso, seus fluxos MCP viram uma “caixa‑preta” unidirecional.

Erro nº 6: eventos de progresso em demasia.
A vontade de “ser honesto” e enviar job.progress a cada um por cento concluído sobrecarrega a rede, o cliente e o servidor MCP. É melhor enviar progresso na mudança de etapa ou em deltas grandes (por exemplo, a cada 10 %), e manter o resto apenas nos logs internos.

Erro nº 7: usar fila em memória em produção.
O exemplo didático com queue: string[] e Map é ótimo para entender a arquitetura, mas em um sistema de produção real ele desaba no primeiro restart do processo ou queda do servidor. Para uso sério, são necessárias filas e armazenamentos externos: SQS, Pub/Sub, RabbitMQ, Redis Streams etc. Soluções em memória servem apenas para desenvolvimento local e demos simples.