TypeScript para engenheiros de ML em Python: criar um serviço de agentes

Viacheslav Dubrov

2026-08-15 · Atualizado 2026-10-02 · 30 min · Ferramentas de desenvolvimento · Agentes de IA

Tradução automática

Este artigo foi traduzido automaticamente a partir da versão original em inglês.

Este guia destina-se a engenheiros de ML e de backend experientes em Python que precisam de criar um serviço de AI em TypeScript e Node. Parte dos conceitos de serviços Python, não da programação básica nem do código para browser. No final, consegue mapear uma stack de serviços Python para TypeScript, reconhecer os hábitos de Python que causam bugs em JavaScript e executar e alterar um serviço de agentes com streaming.

O serviço de exemplo acompanha um pedido ao longo da validação, da execução de ferramentas, do streaming HTTP e da persistência. Responde a três perguntas: o que acontece quando os argumentos de uma ferramenta chegam em fragmentos, quando um stream falha depois de HTTP 200 ou quando um worker reclama o mesmo job duas vezes? O âmbito é apenas backend e trabalho com AI: sem React e sem bundler para browser.

O código está em slavadubrov/typescript-agent-service. Inclui duas implementações do agent loop, o histórico de execuções em Postgres, um worker e um servidor MCP. Pode executar a demonstração offline sem uma chave de API e, depois, usar o mapa de Python para TypeScript abaixo para ler e modificar o serviço. Os cálculos numéricos pesados ficam em Python; este guia cobre orquestração, HTTP e streaming.

Execute primeiro o repositório complementar

Instale o Node 24 e o pnpm seguindo as instruções oficiais de instalação do pnpm. Depois, clone o repositório complementar e execute:

pnpm install
pnpm demo
pnpm check

pnpm demo exercita o handler HTTP, o agent loop e o stream SSE com um modelo simulado por script. Também chama o cálculo runSweep do worker. Não inicia o processo do worker nem a sua fila na base de dados. A demonstração não precisa de chave de API, base de dados nem Docker. pnpm check executa o verificador de tipos, o linter, a verificação do formatter e os testes.

A demonstração imprime as frames SSE de um pedido de chat e, a seguir, o varrimento do worker. Em versão reduzida:

event: tool_call
data: {"type":"tool_call","callId":"call_1","toolName":"estimate_kv_cache","args":{"model":"llama-3.1-8b","seqLen":32768,"batchSize":16}}

event: tool_result
data: {"type":"tool_result","callId":"call_1","toolName":"estimate_kv_cache","ok":true,"result":{"model":"llama-3.1-8b","seqLen":32768,"batchSize":16,"kvCacheGiB":64},"durationMs":0}

event: done
data: {"type":"done","stopReason":"stop","steps":2,"text":"Llama-3.1-8B at 32k context and batch 16 needs 64 GiB of KV cache."}

--- worker: capacity sweep ---
batch=  1  kv_cache=4.0 GiB
batch= 16  kv_cache=64.0 GiB

Cada bloco event:/data: é uma frame SSE: o modelo pede uma ferramenta, a ferramenta responde e a execução termina com done. A última parte é o varrimento de tamanhos de batch do worker.

Para executar a API e o worker reais:

cp .env.example .env          # add OPENAI_API_KEY or an OpenAI-compatible URL
pnpm db:up                    # start Postgres in Docker
pnpm db:push                  # create the database schema
pnpm dev:api                  # API at http://localhost:8080
pnpm dev:worker               # run this in a second terminal

Transmita uma execução do agente. A resposta é uma série de frames SSE como a saída da demonstração acima:

curl -N -X POST localhost:8080/v1/chat \
  -H 'content-type: application/json' \
  -d '{"message":"How much KV cache for llama-3.1-70b at 32k context, batch 8?"}'

Coloque uma tarefa longa na fila e, depois, consulte-a. A primeira chamada devolve 202 com {"id":"...","status":"queued"}. A segunda devolve a linha guardada, com o resultado depois de o worker terminar:

curl -X POST localhost:8080/v1/sweeps \
  -H 'content-type: application/json' \
  -d '{"model":"llama-3.1-70b","seqLen":32768,"batchSizes":[1,4,16]}'

curl localhost:8080/v1/runs/<id>

Os scripts pnpm acima são os comandos que uso no resto deste artigo. O repositório coloca os comandos de nível mais baixo de Node e TypeScript atrás de scripts pnpm com nome, tal como um projeto Python pode colocar comandos uv run atrás de alvos make. Não misture npm install neste repositório pnpm. Use pnpm install para que pnpm-lock.yaml continue a ser o único ficheiro de bloqueio de versões.

O que fazem Node, npm, pnpm, TypeScript e tsx

Os nomes parecidos escondem funções distintas:

• JavaScript é a linguagem.
• Node.js é o runtime, aproximadamente o CPython do JavaScript. Instale o Node 24 para este projeto.
• O registo npm é o índice de pacotes, aproximadamente o PyPI. O comando npm vem com o Node e consegue instalar pacotes desse registo.
• pnpm é o gestor de pacotes que este repositório escolheu. Instala a partir do registo npm, gere o workspace do monorepo e executa os comandos declarados em package.json.
• package.json é o manifesto do projeto, o equivalente mais próximo de pyproject.toml. A sua secção scripts dá nomes como demo, check e dev:api a comandos mais longos.
• TypeScript é JavaScript com tipos estáticos. O seu comando tsc verifica esses tipos. O repositório executa-o através de pnpm check ou pnpm typecheck.
• tsx executa ficheiros .ts sem uma compilação separada. Os scripts de desenvolvimento usam o seu modo watch para reiniciar a API ou o worker depois de uma alteração ao código. Neste guia, não o invoca diretamente.

Neste repositório, execute os scripts pnpm. O Node é o runtime dentro desses scripts, e o Dockerfile fornecido trata da produção.

O que cada script executa:

• pnpm demo executa tsx scripts/demo.ts.
• pnpm dev:api executa tsx watch.
• pnpm check executa tsc, o Biome (o linter) e o Vitest (o executor de testes).
• O CMD do Docker executa node --import tsx/esm.

Todos correm no Node 24. Em pnpm check, o tsc e o Biome apenas leem ficheiros; o Vitest executa os testes no Node.

A stack, mapeada a partir de Python

Zod, Hono, Drizzle, Vitest e Biome cobrem grande parte do mesmo terreno que pydantic, FastAPI, SQLAlchemy, pytest e Ruff.

A maioria dos pares são substituições diretas. Três não são:

O repositório complementar usa quatro bibliotecas que vale a pena explicar.

Hono para a camada HTTP

Express e Fastify são alternativas centradas no Node. O Hono usa as APIs Request e Response padrão da Web e oferece adaptadores para Node e para runtimes serverless. Essa portabilidade é útil para esta pequena API de streaming, por isso escolhi o Hono. Escolha o Fastify quando o serviço corre apenas em Node e quer o seu conjunto de plugins e a serialização de respostas baseada em schemas.

Drizzle para SQL

O Drizzle mantém o schema em TypeScript e não exige um passo de geração de cliente. Também permite usar SQL puro quando o construtor de queries não consegue exprimir bem uma cláusula do Postgres. Eu escolheria antes o Prisma quando o seu cliente gerado e as ferramentas à volta servissem melhor a equipa.

Biome para linting e formatação

O Biome trata de linting, formatação e ordenação de imports com um só binário e um só ficheiro de configuração. Mantenha o ESLint quando o projeto depende de regras personalizadas que o Biome não oferece.

Vitest para testes

O Vitest executa os testes .ts do repositório complementar sem configuração de transformação separada. Mantenha o Jest num repositório que já tem uma configuração Jest e matchers personalizados.

Leia a sintaxe TypeScript usada abaixo

Tenha esta tabela à mão, ao lado dos exemplos do serviço, como referência.

Use const, a menos que a associação tenha de mudar. Use let para um contador, um acumulador ou outra associação que vá reatribuir.

Uma tradução rápida de enum

Para estados com valores em string, este repositório usa um objeto mais um tipo de união de strings inferido:

const Status = { Queued: "queued", Running: "running" } as const;
type Status = (typeof Status)[keyof typeof Status]; // "queued" | "running"

O objeto fornece Status.Queued enquanto o programa corre. A linha type só permite "queued" ou "running" durante a verificação de tipos. Juntos, desempenham os dois papéis desta declaração Python:

from enum import Enum

class Status(str, Enum):
    QUEUED = "queued"
    RUNNING = "running"

Basta reconhecer o padrão. as const mantém os valores do objeto como as strings exatas, em vez de os alargar para um string qualquer.

As sete diferenças semânticas que custam tempo

A tabela de sintaxe leva-o até ao fim dos exemplos. É nestas diferenças semânticas que os hábitos de Python causam bugs.

1. Arrays e objetos vazios são truthy

O “contentor vazio é falsy” de Python é o hábito que pior se transfere. if (results) é verdadeiro para um array vazio. Escreva if (results.length).

2. null e undefined são diferentes

null costuma marcar uma ausência deliberada. undefined costuma significar que um valor falta ou não foi atribuído, embora o código o possa atribuir explicitamente. O código de bibliotecas devolve undefined a toda a hora. A diferença aparece quando escreve um valor por omissão. || substitui pelo lado direito sempre que o esquerdo é falsy. Isso inclui 0, "" e false. ?? só substitui para null e undefined. Assim, 0 || 10 é 10, enquanto 0 ?? 10 é 0. É essa diferença que faz um tamanho de batch igual a zero passar em silêncio a dez.

3. Um bloco catch recebe unknown

Não existe except ValueError:. Um único bloco catch recebe tudo. Como o JavaScript permite lançar uma string, um número ou null, o TypeScript atribui ao valor capturado o tipo unknown, o seu tipo “pode ser literalmente qualquer coisa” em modo strict. O repositório complementar ativa strict, e os projetos novos devem, em geral, fazê-lo. Para inspecionar o erro, restrinja primeiro o valor:

try {
    await risky();
} catch (error) {
    // `error` is `unknown` until you prove otherwise. This line is the
    // TypeScript equivalent of `except Exception as e:` and it is not
    // optional.
    const message = error instanceof Error ? error.message : String(error);
}

4. As promises começam logo

Chamar uma função async começa a executar o seu corpo e devolve uma promise. Um objeto de corrotina Python não faz nada até ser aguardado ou agendado. Promise.all aproxima-se de asyncio.gather. Promise.allSettled aproxima-se de gather(..., return_exceptions=True), exceto que cada resultado vem embrulhado como { status, value } ou { status, reason }.

O Node é responsável pelo agendamento em runtime. Mantém o processo vivo enquanto existirem handles ou pedidos ativos, como temporizadores e sockets. Uma simples promise pendente não mantém o Node vivo. Não embrulha o programa em asyncio.run. Num módulo ES, pode usar await ao nível superior quando o arranque tem de esperar por uma operação assíncrona.

5. O JavaScript tem um único tipo numérico comum

O tipo number do JavaScript guarda os valores como números de vírgula flutuante de 64 bits, aproximadamente o mesmo que o float de Python. A norma técnica deste formato chama-se IEEE 754. Os valores decimais são aproximados, por isso 0.1 + 0.2 não é exatamente 0.3, e os inteiros só são exatos até 2**53 - 1, ou 9,007,199,254,740,991.

Mantenha os IDs de 64 bits como strings nas fronteiras do serviço. Converter um bigint do Postgres para um number do JavaScript pode arredondá-lo. Para inteiros exatos maiores, o JavaScript oferece o tipo BigInt, à parte, que não se mistura com os números comuns.

6. Use Map quando precisar de um dicionário ao estilo Python

Em JavaScript, {} cria um objeto. Os objetos costumam representar registos com campos nomeados:

const request = { model: "gpt-5", seqLen: 4096 };
console.log(request.model);

Um objeto não é uma tabela de chave-valor limpa como um dict de Python. Herda alguns nomes do próprio JavaScript. Isto pode dar um resultado surpreendente:

const tools: Record<string, unknown> = {};

tools["constructor"]; // a built-in function, not a missing value
Object.hasOwn(tools, "constructor"); // false

Se uma string externa seleciona um campo de um objeto, chame Object.hasOwn antes de o ler. Se precisar de um dicionário de uso geral, use Map. O Map está mais perto do dict de Python: uma chave só existe quando o seu código a adiciona.

const tools = new Map<string, unknown>();
tools.get("constructor"); // undefined

7. Inclua a extensão nos imports relativos

Um ficheiro de código JavaScript que partilha código com outros ficheiros chama-se módulo. Este projeto usa o formato de módulo moderno, os módulos ES, normalmente abreviado para ESM. ES significa ECMAScript, o nome formal da linguagem JavaScript. Na prática, o ESM é a sintaxe import e export usada em todo este projeto.

Para um import relativo, o Node exige o nome exato do ficheiro. Não adivinha se ./env quer dizer ./env.ts ou ./env.js:

import { loadEnv } from "./env.ts";

Os imports de pacotes instalados ou do workspace continuam a usar o nome do pacote, sem extensão de ficheiro:

import { z } from "zod";
import { runAgent } from "@agent/core";

Zod: declare o schema, derive o tipo

Em pydantic, declara-se uma classe e obtém-se um validador. Em Zod, declara-se um validador e deriva-se o tipo a partir dele. É a mesma fonte única de verdade, na direção oposta.

A figura mostra a mesma definição PORT nos dois sentidos, e o que resta de cada declaração enquanto o código corre.

De packages/schemas/src/env.ts:

import { z } from "zod"; // `z` is Zod's whole API, the way `pd` is pandas

const EnvSchema = z.object({
    PORT: z.coerce.number().int().positive().default(8080),
    // See the note below: a bare z.url() would accept "localhost:8000".
    OPENAI_BASE_URL: z
        .url({ protocol: /^https?$/ })
        .default("https://api.openai.com/v1"),
    DATABASE_URL: z.string().optional(),
});

export type Env = z.infer<typeof EnvSchema>;

z.infer<typeof EnvSchema> lê um tipo estático a partir do schema de runtime. z.coerce.number() trata do facto de cada valor definido em process.env (o os.environ do Node) ser uma string. Desempenha o mesmo papel que a coerção de definições numéricas do pydantic, embora as strings exatas que cada um aceita sejam diferentes. Segue o padrão do pydantic-settings e corre uma vez no arranque. Um ambiente inválido produz então um erro de arranque legível, em vez de um TypeError dentro de um handler.

Um z.url() simples aceita localhost:8000. A norma de URL trata tudo o que vem antes dos primeiros dois pontos como o esquema. Por isso lê localhost: como um protocolo chamado “localhost” e aceita a string. O valor chega então ao cliente HTTP e falha com menos contexto. A validação de schemas antecipa as falhas, mas impõe um schema permissivo se for isso que escreveu.

O Zod 4 também inclui z.toJSONSchema, pelo que este projeto não precisa da dependência zod-to-json-schema comum em tutoriais mais antigos. Isso importa quando um só schema tem de alimentar três consumidores, que é o tema da secção “Uma ferramenta, três consumidores” mais abaixo.

O serviço

A linha de cima é código partilhado em packages/. Por baixo, cada aplicação em apps/ é um processo separado. Só a API e o worker acedem ao Postgres, e o servidor MCP fala com o seu cliente por stdin e stdout.

O serviço de demonstração dimensiona deployments de LLM. Uma ferramenta consulta as constantes de arquitetura de um modelo. A outra estima a sua ocupação de KV cache: a memória da GPU usada para guardar as chaves e os valores de atenção dos pedidos em curso. As duas ferramentas são deliberadamente aritmética aborrecida. Não precisam de rede e dão sempre a mesma resposta. Isso torna o serviço testável sem chave de API. O estimador de KV cache também é publicado através do Model Context Protocol (MCP), para que outros clientes de AI o possam chamar.

typescript-agent-service/
├── apps/
│   ├── api/           @agent/api: Hono API, streams Server-Sent Events (below)
│   ├── worker/        polls Postgres for long-running jobs
│   └── mcp/           MCP server: exposes one tool to outside AI clients
├── packages/
│   ├── schemas/       package name @agent/schemas: env, API, and tool schemas
│   ├── agent-core/    package name @agent/core: the loop (twice), tools, storage
│   └── observability/ Pino logging, OpenTelemetry tracing
├── pnpm-workspace.yaml
└── package.json

pnpm-workspace.yaml é o ficheiro que declara o workspace. Os pacotes internos recebem um nome com âmbito, como @agent/core, em que o prefixo @agent/ é uma convenção de nomes, não uma funcionalidade da linguagem. Cada pacote declara o seu ponto de entrada público em package.json. Essa fronteira do pacote não depende do comando que arranca a aplicação.

Este workspace privado aponta essas entradas para código .ts, porque todos os consumidores fazem parte do mesmo repositório. Os pacotes npm públicos costumam publicar JavaScript mais declarações de tipos .d.ts, para que os consumidores comuns de Node não precisem do executor de TypeScript nem da configuração de build do autor do pacote.

Escreva o loop de ferramentas à mão, uma vez

A coluna da esquerda é um passo do modelo. A coluna da direita é o que o loop emite: exceto uma exceção que sai do loop, esses eventos são tudo o que a rota HTTP e os testes veem.

Os frameworks de agentes com tool calls envolvem o mesmo loop básico:

1. Chamar o modelo com as definições das ferramentas.
2. Validar e executar as ferramentas pedidas.
3. Acrescentar os resultados às mensagens.
4. Chamar o modelo outra vez.

Escrever o loop uma vez torna visíveis os seus caminhos de falha escondidos.

O loop é uma async function*, um gerador assíncrono, exatamente a forma do async def de Python com yield. A rota HTTP itera esse gerador, transforma cada evento numa frame de Server-Sent Events e acumula o texto. Depois de o stream terminar, a rota chama storage.createRun uma vez com o texto final. Os testes invocam o loop em separado e recolhem os seus eventos num array. Uma frame SSE é um fragmento de uma resposta HTTP de longa duração. A secção abaixo explica o formato.

O loop tem duas saídas normais. Uma resposta sem tool calls emite done com stopReason: "stop". Se o modelo continuar a pedir ferramentas até maxSteps, o gerador conclui na mesma e emite done com stopReason: "max_steps". Trate esse resultado como concluído mas truncado: o texto acumulado pode ser parcial.

Cada valor que o loop emite é um AgentEvent, uma união etiquetada. Em versão reduzida, de packages/schemas/src/events.ts:

export const AgentEventSchema = z.discriminatedUnion("type", [
    z.object({ type: z.literal("text"), delta: z.string() }),
    z.object({
        type: z.literal("tool_call"),
        callId: z.string(),
        toolName: z.string(),
        args: z.unknown(),
    }),
    z.object({
        type: z.literal("tool_result"),
        callId: z.string(),
        ok: z.boolean(),
        result: z.unknown() /* ... */,
    }),
    z.object({
        type: z.literal("done"),
        stopReason: z.enum(["stop", "max_steps"]),
        steps: z.number(),
        text: z.string(),
    }),
    z.object({ type: z.literal("error"), message: z.string() }),
]);

export type AgentEvent = z.infer<typeof AgentEventSchema>;

z.discriminatedUnion escolhe a variante pelo campo type, como um match sobre dataclasses identificadas por uma etiqueta. Dentro de if (event.type === "text"), o compilador sabe que event.delta existe. Adicionar uma variante não faz um switch incompleto falhar automaticamente. Para obter um erro de compilação quando falta uma variante, verifique o valor restante contra never num ramo default. Uma regra de lint para verificar todos os casos também pode detetar os que faltam. A rota HTTP e a versão com o AI SDK abaixo utilizam os tipos de evento restringidos.

O corpo do loop emite eventos de texto e monta as tool calls a partir de fragmentos. De packages/agent-core/src/loop.ts:

let text = ""; // the assistant text produced during this model step
// Keyed by the `index` field, because a streamed response interleaves
// fragments of several parallel tool calls and only `index` is present
// on every fragment. `id` and `name` arrive once, `arguments` arrives
// in pieces. `a?.b` below reads `b` only if `a` exists, and gives back
// `undefined` instead of throwing if it doesn't.
const partial = new Map<number, PartialToolCall>();

for await (const chunk of stream) {
    const choice = chunk.choices[0];
    if (!choice) continue;

    if (choice.delta.content) {
        text += choice.delta.content;
        yield { type: "text", delta: choice.delta.content };
    }

    for (const fragment of choice.delta.tool_calls ?? []) {
        const slot = partial.get(fragment.index) ?? { id: "", name: "", args: "" };
        if (fragment.id) slot.id = fragment.id;
        if (fragment.function?.name) slot.name = fragment.function.name;
        if (fragment.function?.arguments) slot.args += fragment.function.arguments;
        partial.set(fragment.index, slot);
    }
}

A variável text, local ao loop, guarda um passo do modelo. O loop usa-a na mensagem do assistente para o passo seguinte ou no evento done final. Não é o acumulador ao nível da rota que depois é persistido.

O mapa partial é a parte que os frameworks escondem. O SDK expõe fragmentos de string dos argumentos da função, e esses fragmentos podem cortar o JSON serializado em posições arbitrárias. Várias chamadas paralelas também podem intercalar-se. O repositório tem um teste que divide {"model":"llama-3.1-8b",...} em quatro chunks.

A figura segue esses quatro chunks até à entrada do mapa para o índice 0.

A segunda coisa que vale a pena escrever você mesmo é o que acontece quando a validação falha:

// `tool.parse` is this repo's wrapper, not Zod's. Zod's own `.parse` throws and
// `.safeParse` returns `{ success, data, error }`; this returns
// `{ ok: true, value }` on success and `{ ok: false, error }` on failure, so no
// caller has to catch.
const parsed = tool.parse(raw);
if (!parsed.ok) {
    return {
        ok: false,
        value: { error: `Invalid arguments: ${parsed.error}` },
        parsedArgs: raw,
    };
}

Antes da execução, a procura pode não encontrar a ferramenta, o JSON.parse pode rejeitar os argumentos, ou o Zod pode rejeitar a sua forma. Cada falha torna-se uma mensagem que o modelo lê. Durante a execução, um ToolError esperado também se torna um tool result, para que o modelo possa corrigir a chamada. Uma exceção inesperada propaga-se para o caminho de erro HTTP, em vez de ser apresentada como falha de domínio. z.prettifyError transforma a árvore de problemas do Zod numa mensagem em que o modelo pode atuar, em vez de um stack trace.

strict: true numa definição de função da OpenAI pede ao fornecedor que restrinja a descodificação ao schema. Não tem relação com a flag strict do tsconfig do TypeScript. Assemelha-se à guided decoding do vLLM, embora os schemas suportados e os detalhes de aplicação sejam diferentes. Elimina um modo de falha, mas um endpoint auto-alojado pode ignorar a flag. Os argumentos também têm de passar pelo JSON.parse.

O loop chama /chat/completions porque o repositório complementar tem como alvo servidores compatíveis com a OpenAI. O vLLM, o SGLang e o Ollama documentam esse endpoint, pelo que OPENAI_BASE_URL pode apontar o mesmo cliente a qualquer um deles. A cobertura da Responses API difere e muda de versão para versão. Se controla os dois lados, consulte a página de compatibilidade atual do servidor antes de escolher entre as duas APIs.

Depois mude para o AI SDK e saiba o que trocou

O repositório complementar também implementa o mesmo agente com o Vercel AI SDK, para comparar as duas versões. Ambas produzem valores AgentEvent, mas o comportamento perante erros terminais difere atualmente. O adaptador opcional do SDK produz um evento error com String(part.error) e termina sem done. A rota HTTP atual reencaminha esse detalhe e guarda a execução como succeeded, porque o gerador terminou sem lançar uma exceção. Antes de substituir o loop direto predefinido por este adaptador, trate error ou um stream que feche sem done como uma execução falhada, remova detalhes sensíveis das mensagens de erro para o cliente e teste o estado guardado juntamente com o evento terminal.

O repositório complementar usa o AI SDK 7. A sua implementação com o framework está em packages/agent-core/src/loop-ai-sdk.ts:

const result = streamText({
    model: provider.chatModel(options.model),
    prompt: options.message,
    tools: aiSdkTools,          // Zod schemas passed straight through
    stopWhen: stepCountIs(options.maxSteps ?? 6),
});

for await (const part of result.fullStream) {
    switch (part.type) {
        case "text-delta": yield { type: "text", delta: part.text }; break;
        case "tool-call":  yield { type: "tool_call", callId: part.toolCallId, /* ... */ }; break;
        // ...
    }
}

O SDK elimina cinco partes do código da aplicação:

• o acumulador de fragmentos
• JSON.parse e o seu caminho de erro
• a chamada que executa o Zod sobre os argumentos analisados
• a montagem de mensagens específica de cada fornecedor
• o contador de passos

stopWhen aceita várias condições, incluindo um limite de passos ou uma tool call específica. O loop for await não muda quando a política de paragem muda.

O que se perde é o controlo direto sobre a falha de validação. O loop escrito à mão decide o que o modelo vê depois de uma chamada rejeitada. O repositório complementar não define a opção repairToolCall do SDK, por isso usa o comportamento por omissão: com mais de um passo permitido, o AI SDK 7 devolve a chamada falhada ao modelo no passo seguinte. Passe um callback repairToolCall para corrigir a chamada você mesmo, por exemplo com uma segunda chamada ao modelo.

A troca também funciona no sentido contrário. Na versão com o SDK, uma mudança de fornecedor fica localizada no adaptador do fornecedor. Continua a exigir o pacote do fornecedor correspondente, as credenciais, a configuração e testes de integração. No loop em bruto, o tratamento de pedidos e de streams específico de cada fornecedor é código seu para alterar.

Para um serviço novo, comece com o SDK, a menos que precise de controlar esses caminhos de falha. A versão escrita à mão dá-lhe uma implementação pequena para inspecionar e comparar com essa escolha.

Streaming por HTTP: Hono e SSE

As rotas do Hono leem-se como as rotas do FastAPI. A única adição é o zValidator, que faz o trabalho que o FastAPI obtém de graça a partir das anotações de tipo na assinatura de um handler. O c no handler abaixo é o contexto de pedido do Hono, o objeto que o FastAPI reparte pelos seus parâmetros. deps é um conjunto de dependências com que a app é construída, em vez de as importar diretamente. runAgent é uma delas, e a secção de testes mostra o que isso lhe dá.

Em versão reduzida, de apps/api/src/app.ts:

app.post("/v1/chat", zValidator("json", ChatRequestSchema), (c) => {
    const body = c.req.valid("json");
    const log = deps.logger.child({ route: "chat" });

    return streamSSE(c, async (stream) => {
        let text = "";
        let done: Extract<AgentEvent, { type: "done" }> | undefined;
        try {
            // The full route wraps this loop in a `withSpan` tracing span.
            for await (const event of deps.runAgent({
                message: body.message,
                maxSteps: body.maxSteps,
            })) {
                if (event.type === "text") text += event.delta;
                if (event.type === "done") done = event;

                await stream.writeSSE({
                    event: event.type,
                    data: JSON.stringify(event),
                });
            }
        } catch (error) {
            log.error({ err: error }, "agent run failed");
            await stream.writeSSE({
                event: "error",
                data: JSON.stringify({
                    type: "error",
                    message: "Agent run failed",
                }),
            });
            return;
        }

        await deps.storage.createRun({
            kind: "chat",
            status:
                done?.stopReason === "max_steps" ? "truncated" : "succeeded",
            input: { message: body.message },
            output: { text },
        });
    });
});

A rota guarda o evento done final. Um loop que pára em max_steps conclui normalmente, por isso a rota guarda-o como truncated, e não como succeeded.

zValidator valida o corpo e dá a c.req.valid("json") o tipo que o schema produz. Sem ele, o corpo fica com o tipo any, a saída de emergência do TypeScript, em que todos os acessos a propriedades compilam e nada é verificado. Isso anula o benefício de segurança de tipos do schema.

Esta rota usa Server-Sent Events em vez de WebSockets. O servidor mantém uma resposta HTTP aberta enquanto escreve frames event: <name> e data: <json> e fecha-a depois do evento final. O tráfego flui do servidor para o cliente, o que condiz com este stream de agente. Um WebSocket acrescentaria mensagens bidirecionais e uma atualização de protocolo de que esta rota não precisa.

Com o loop direto predefinido, uma exceção a meio do stream altera a comunicação de erros HTTP. Depois de enviar o primeiro frame com estado 200, o servidor não pode substituir essa resposta por um 500. O bloco catch regista o erro capturado, envia um evento de erro constante ao cliente e termina. Esse retorno omite storage.createRun, pelo que um erro lançado pelo gerador não deixa nenhuma linha. Este catch não é executado quando o adaptador opcional do SDK produz um evento error e termina. Para guardar execuções falhadas para depuração ou avaliação, escreva uma linha failed no bloco catch com o texto parcial.

A figura segue um pedido até aos seus três finais. O código de estado é o mesmo nos três, por isso só a última frame os distingue.

Um teste cobre este caminho. Um gerador emite um delta de texto e depois lança uma exceção. A resposta continua a ser 200, e a sua última frame é um evento error com a mensagem constante Agent run failed. O logger guarda o erro capturado para diagnóstico no servidor. Qualquer cliente que verifique apenas o código de estado dá uma execução falhada como bem-sucedida.

O app.ts toma duas decisões mais pequenas que merecem explicação. Trata /healthz como um endpoint de liveness, por isso essa rota deliberadamente não toca no Postgres. Uma falha de liveness durante uma indisponibilidade da base de dados poderia reiniciar todas as réplicas sem reparar a dependência. Acrescente uma verificação de readiness separada quando o orquestrador tiver de parar de encaminhar tráfego para uma instância que não consegue chegar ao Postgres. Os caminhos de erro registam o erro capturado, mas devolvem uma string constante. Devolver error.message no corpo de uma resposta é a forma de as strings de ligação acabarem no browser de outra pessoa.

A parte à Celery, sem Celery

As tarefas longas não pertencem a um handler de pedidos. POST /v1/sweeps valida um modelo, um comprimento de sequência e uma lista de tamanhos de batch, insere uma linha queued e devolve 202 com o id da execução. O worker reclama a linha, calcula o tamanho da KV cache para cada tamanho de batch e guarda o resultado. GET /v1/runs/:id lê-o.

A metade de cima mostra os estados por que passa uma linha de varrimento. A linha temporal mostra a recuperação de bloqueio expirado explicada abaixo: passados cinco minutos, um segundo worker reclama uma tarefa que ainda está a correr.

Aqui não há Redis nem BullMQ. O PostgreSQL documenta SKIP LOCKED para vários consumidores de uma tabela semelhante a uma fila. A cláusula dá a este pequeno serviço uma fila at-least-once numa só tabela. É transacional com o resto das suas escritas e é um serviço a menos no docker-compose.yml.

A query de reclamação em packages/agent-core/src/db/storage.ts é:

const [candidate] = await tx
    .select({ id: runs.id })
    .from(runs)
    // The real query also picks up rows whose lock went stale; trimmed here.
    .where(and(eq(runs.kind, kind), eq(runs.status, "queued")))
    .orderBy(runs.createdAt)
    .limit(1)
    // `.for()` adds a locking clause; `{ skipLocked: true }` adds SKIP LOCKED.
    .for("update", { skipLocked: true });

A linha fica bloqueada durante a transação, e qualquer worker concorrente que execute a mesma query salta-a em vez de ficar à espera. Duas reclamações simultâneas não recebem, por isso, a mesma linha não expirada. Um teste de integração dispara duas reclamações ao mesmo tempo com Promise.all e verifica que devolvem linhas diferentes. A versão ingénua, SELECT ... LIMIT 1 seguido de UPDATE, falha esse teste: as duas transações leem a mesma linha antes de qualquer uma escrever, pelo que ambas iniciam a mesma tarefa.

O painel de cima é a condição de corrida que o teste apanha; o painel de baixo é a query de reclamação acima.

Isto é execução at-least-once: uma tarefa pode correr mais de uma vez. A query completa também reclama uma linha running quando o seu bloqueio tem mais de cinco minutos, e nada renova essa concessão. Uma tarefa ativa que corra mais de cinco minutos pode, por isso, ser reclamada duas vezes. Torne as tarefas idempotentes. Para trabalho de longa duração, acrescente um heartbeat da concessão ou defina o limite de bloqueio expirado acima do tempo de execução máximo.

Acrescente o BullMQ quando precisar de tarefas com atraso, agendamentos repetíveis, prioridades, limites de taxa ou um painel. O passo equivalente em Python é ir de uma tabela da base de dados para o Celery. Antes disso, o Redis é mais um serviço para executar e monitorizar.

O worker volta a validar o que lê de jsonb:

// The row was validated on the way in, but it has been through a database.
// A stored row can outlive the schema version that accepted it.
// The job is a batch-size sweep. Zod's own `.parse` throws; the poll loop
// catches that and marks the run failed.
const input = SweepRequestSchema.parse(run.input);

A suite de testes também alimenta o worker com uma linha de varrimento cujo seqLen é a string "lots". O worker marca a execução como falhada e continua a consultar, em vez de falhar e voltar a tentar a mesma linha inválida para sempre.

O trabalho de CPU expõe outra restrição do Node. Um callback síncrono corre na thread do event loop e não é interrompido. Um ciclo for a fazer aritmética durante dois segundos bloqueia, durante dois segundos, todos os pedidos, temporizadores e verificações de liveness desse processo. Um ciclo apertado dentro de um async def bloqueia o asyncio da mesma forma. Os dois runtimes exigem que delegue explicitamente o trabalho de CPU.

await setTimeout(0) de node:timers/promises (o prefixo node: significa biblioteca padrão, pelo que node:timers está para o Node como os está para o Python) é await asyncio.sleep(0). O varrimento cede o controlo depois de cada tamanho de batch, para que o processo do worker possa atender temporizadores e outros callbacks. Ceder o controlo não torna o trabalho de CPU paralelo. As worker_threads do Node conseguem executar JavaScript em paralelo. Para trabalho de CPU em Python puro com a build habitual do CPython com GIL, use um pool de processos em vez de um pool de threads. Este serviço não usa nenhum dos dois. Mantenha os cálculos numéricos pesados em Python, onde as bibliotecas de suporte já estão, e retire-os do event loop da API.

Uma ferramenta, três consumidores

EstimateKvCacheInput tem três consumidores:

• O loop escrito à mão converte-o com z.toJSONSchema.
• O AI SDK recebe-o sem alterações.
• O servidor MCP publica a sua forma.

Essa reutilização é a razão de packages/schemas existir.

De apps/mcp/src/index.ts:

server.registerTool(
    "estimate_kv_cache",
    {
        description: "Estimate KV-cache VRAM in GiB for a served model...",
        // A Zod object schema keeps the map of fields you passed in on
        // `.shape`. This SDK wants that map, not the schema wrapped around it.
        inputSchema: EstimateKvCacheInput.shape,
    },
    async ({ model, seqLen, batchSize }) => {
        /* ... */
    },
);

await server.connect(new StdioServerTransport());

Há dois detalhes que importam antes de ligar um cliente. Primeiro, um servidor iniciado desta forma usa o seu próprio stdin e stdout para comunicar com o cliente. Cada linha é uma mensagem JSON-RPC. Um console.log perdido, o equivalente em JavaScript de print, corrompe então uma mensagem. O cliente desliga-se com um erro de análise que não indica nenhum ficheiro. Envie antes todos os diagnósticos para o stderr.

Segundo, uma falha de domínio deve devolver isError: true com uma mensagem. O modelo que chama pode então corrigir a chamada, tal como pode depois de argumentos de ferramenta inválidos no agent loop.

Pode comandar o servidor com printf e um pipe, o que vale a pena fazer uma vez antes de apontar um cliente real para ele:

printf '%s\n' \
  '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"probe","version":"1"}}}' \
  '{"jsonrpc":"2.0","method":"notifications/initialized"}' \
  '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"estimate_kv_cache","arguments":{"model":"llama-3.1-70b","seqLen":8192,"batchSize":4}}}' \
  | pnpm -s mcp   # -s suppresses pnpm's own output so only JSON-RPC comes back

A sonda corresponde à versão do protocolo usada no README do repositório complementar. Para um cliente real, use o SDK em vez de manter mensagens JSON-RPC à mão.

Testar um agente sem chave de API

O Vitest desempenha o papel do pytest, mas a estrutura é diferente. describe agrupa testes relacionados. it e test definem cada um um caso de teste. test.each aproxima-se de parametrize, beforeEach fornece a preparação de cada teste, vi.fn() cria uma função mock e describe.skipIf ignora um grupo de forma condicional.

Os testes do agente dependem de uma decisão: runAgent recebe um cliente OpenAI como parâmetro em vez de construir um. O falso é um objeto com um método chat.completions.create que devolve um iterável assíncrono simulado por script:

function fakeClient(scripts: Chunk[][]): OpenAI {
    let call = 0;
    return {
        chat: {
            completions: {
                create: async () => {
                    const script = scripts[call++] ?? [];
                    // Defines an async generator and calls it on the same
                    // line, so `create` hands back something you can
                    // `for await` over, which is the shape a real streaming
                    // response has.
                    return (async function* () {
                        for (const chunk of script) yield chunk;
                    })();
                },
            },
        },
        // TypeScript refuses a direct cast between unrelated shapes, so you
        // cast through `unknown` first. This disables type checking for that
        // value, so keep it to test code.
    } as unknown as OpenAI;
}

Os testes dividem uma string de argumentos JSON por vários chunks e tratam duas tool calls numa só resposta. Também cobrem um batchSize inválido, JSON malformado, nomes de ferramenta desconhecidos e um modelo que continua a chamar ferramentas até o maxSteps o parar. O ficheiro de testes corre em bem menos de um segundo, sem rede e sem chave.

Os testes de integração com o Postgres usam describe.skipIf(!process.env.DATABASE_URL), pelo que pnpm test funciona num clone novo sem nenhum Postgres a correr, e a CI ativa-os ao fornecer a variável. O repositório tem 42 testes. Trinta e oito correm sem Postgres nem chave de API.

Registe eventos estruturados com Pino

O Pino desempenha o mesmo papel que o structlog: um objeto JSON por linha, loggers filhos com campos associados e ocultação explícita de dados sensíveis. O repositório complementar configura-o em packages/observability/src/logger.ts:

const log = pino({
    redact: {
        paths: [
            "req.headers.authorization",
            "apiKey",
            "OPENAI_API_KEY",
            "*.apiKey",
        ],
        censor: "[redacted]",
    },
});

Sem essa ocultação, log.info({ req }, "...") pode copiar um cabeçalho Authorization para o backend de logs.

Rastreie o trabalho da aplicação com spans manuais

O repositório complementar usa OpenTelemetry para três spans ao nível da aplicação: agent.run, agent.tool e worker.sweep. Não instala instrumentação automática de HTTP nem de Postgres. startTracing() em packages/observability/src/tracing.ts cria um NodeSDK com um exportador de traces OTLP. Se OTEL_EXPORTER_OTLP_ENDPOINT não estiver definido, deixa o tracing desligado.

O próprio trabalho é envolvido por withSpan(), do mesmo ficheiro:

return tracer.startActiveSpan(name, { attributes }, async (span) => {
    try {
        return await fn(span);
    } finally {
        span.end();
    }
});

O JavaScript não tem sintaxe de gestor de contexto ao estilo de Python. Aqui, o callback é o bloco que um gestor de contexto de Python envolveria. O helper completo também regista exceções e define o estado do span antes de voltar a lançar a exceção.

Os spans automáticos de HTTP e de base de dados são uma funcionalidade à parte. Exigem os pacotes de instrumentação correspondentes e a inicialização antes de os módulos instrumentados serem carregados. Acrescente isso apenas quando esses spans forem úteis e siga depois a configuração do SDK de Node do OpenTelemetry para as versões exatas dos pacotes que colocar em produção.

Entregue o monorepo em Docker

Use o Dockerfile fornecido. O contentor arranca a API com o loader tsx. Não precisa de escolher nem de invocar um executor de TypeScript quando o implanta.

A build usa pnpm fetch para que as transferências de dependências fiquem em cache até o lockfile mudar. Depois usa pnpm deploy para copiar a API e as suas dependências de produção para uma pasta autónoma. A fase de runtime corre como o utilizador node, sem privilégios de root, e o seu CMD em forma exec deixa a API receber SIGTERM diretamente, para um encerramento ordenado.

Um percurso de três semanas

Os engenheiros de Python experientes podem saltar o material que ensina variáveis e ciclos. Esta sequência concentra-se nas partes que diferem de Python. Faça a tarefa de construção de cada linha; a leitura serve de apoio.

Comece pelos tutoriais gratuitos do Total TypeScript. Pague pelo material avançado apenas quando trabalhar com genéricos e tipos condicionais ao nível de bibliotecas. Salte todos os cursos de “introdução ao JavaScript” e salte tudo o que se pareça com React, a menos que o produto o exija.

Para um panorama das convenções de produção, goldbergyoni/nodebestpractices é uma lista de verificação ampla mantida pela comunidade. Confirme na documentação atual do Node os conselhos que afetem o comportamento em runtime ou a segurança.

Antes de adaptar o serviço

Os testes offline exercitam o fluxo de controlo do serviço com respostas de modelo simuladas por script. Não estabelecem com que fiabilidade um modelo real concluirá as suas tarefas. Mantenha essa distinção quando substituir as ferramentas de demonstração por ferramentas reais.

Valide cada valor que entra de fora: o ambiente, o corpo HTTP, os argumentos de tool call do modelo e as linhas lidas de jsonb.

Defina a seguir a política de conclusão: decida se a sua aplicação aceita, volta a tentar ou revê as execuções truncated, e teste em conjunto o evento e o estado guardado. Se as tarefas puderem correr mais tempo do que o limite de bloqueio expirado, acrescente um heartbeat da concessão antes de acrescentar efeitos secundários.

Quando esses comportamentos estiverem explícitos, acrescente a linha failed descrita na secção de streaming e use as execuções falhadas e truncadas para construir uma suite de avaliação de agentes separada. É aí que se testa a conclusão de tarefas; a suite offline continua a ser a verificação mais rápida de falhas de streaming, de validação e de persistência.

Referências

Repositório de demonstração

• slavadubrov/typescript-agent-service - o monorepo usado ao longo deste artigo: API Hono com SSE, duas implementações do agent loop, armazenamento com Drizzle, worker, servidor MCP e 42 testes

Runtime e linguagem

• Executar TypeScript diretamente em Node.js - o suporte limitado de TypeScript no Node e a sua restrição em node_modules
• Opções do compilador TypeScript - strict e as outras verificações configuradas pelo repositório complementar
• Guia de JavaScript da MDN - a referência da linguagem que vale a pena ter aberta
• javascript.info - tutorial de JavaScript moderno. Leia os capítulos sobre módulos e promises.

Ferramentas

• pnpm e instalação do pnpm - gestor de pacotes, workspaces e configuração
• Biome - linting, formatação e ordenação de imports num só binário
• Vitest - executor de testes que não precisa de configuração de transformação
• Total TypeScript - tutoriais gratuitos e um percurso pago para tipos avançados

Bibliotecas

• Zod - validação de schemas e inferência de tipos. A versão 4 inclui z.toJSONSchema.
• Hono - framework HTTP baseado em normas da Web
• Drizzle ORM - ORM de TypeScript centrado em SQL com migrações drizzle-kit
• Documentação do PostgreSQL sobre SELECT - a cláusula de bloqueio FOR UPDATE ... SKIP LOCKED
• BullMQ - fila baseada em Redis para quando a tabela da base de dados não chega

AI e agentes

• Vercel AI SDK - streamText, tool, stopWhen e adaptadores de fornecedores
• openai/openai-node - o cliente oficial de TypeScript
• MCP TypeScript SDK e a especificação MCP - construção de servidores e clientes

Convenções

• goldbergyoni/nodebestpractices - lista de verificação mantida pela comunidade para convenções de produção