TypeScript für Python-ML-Engineers: Einen Agent Service entwickeln

Viacheslav Dubrov

2026-08-15 · Aktualisiert 2026-08-15 · 22 Min. · Entwicklungswerkzeuge · KI-Agenten

Automatische Übersetzung Dieser Artikel wurde automatisch aus der englischen Originalversion übersetzt.

Dies ist ein kompakter Onboarding-Leitfaden für erfahrene Python-Engineers, die AI-Services in TypeScript und Node ausliefern müssen. Er richtet sich an ML-Engineers, Data Scientists und Backend-Entwickler, die keinen JavaScript-Einsteigerkurs benötigen.

Ich habe dieses Onboarding in den vergangenen Monaten selbst durchlaufen, nachdem ich mit Python und Java gearbeitet hatte. Die meisten Leitfäden, die ich gefunden habe, begannen mit grundlegender Programmierung oder Frontend-DOM-Arbeit. Dieser Artikel geht von den Konzepten eines Python-Service-Stacks aus. Am Ende kannst du einen Python-Service-Stack auf TypeScript abbilden und Python-Gewohnheiten erkennen, die JavaScript-Bugs verursachen. Das durchgehende Beispiel folgt einem Streaming-Agent-Service – vom Schema bis zum Deployment.

Zusammenfassung: Installiere Node 24 und pnpm. Verwende anschließend die pnpm-Commands des Repositorys. Führe pnpm demo für das Offline-Beispiel aus, pnpm dev:api und pnpm dev:worker für die Entwicklung und pnpm check vor einem Commit. Du musst node, tsx oder den TypeScript-Checker nicht selbst ausführen. Das erledigen die Package-Scripts.

Der Service verwendet Zod, Hono, Drizzle, Vitest und Biome. Sie decken einen Großteil derselben Aufgaben ab wie pydantic, FastAPI, SQLAlchemy, pytest und Ruff. Umfangreiche numerische Berechnungen gehören nach Python. Verwende diesen TypeScript-Service für Orchestration, HTTP und Streaming.

Alles hier ist eine Datei aus slavadubrov/typescript-agent-service, dem zu diesem Artikel veröffentlichten Begleit-Repository. Es enthält eine HTTP API, zwei Versionen desselben Agent Loops, Postgres Run History, einen Worker und einen MCP-Server. pnpm install && pnpm demo führt den Offline-HTTP/SSE-Agent-Pfad und die Sweep-Berechnung des Workers ohne API-Key aus.

Ich behandle ausschließlich Backend- und AI-Themen. Kein React. Es gibt auch keinen Browser-Bundler.

Zuerst das Begleit-Repository ausführen

Installiere Node 24 und pnpm gemäß den offiziellen pnpm-Installationsanweisungen. Klone anschließend das Begleit-Repository und führe Folgendes aus:

pnpm install
pnpm demo
pnpm check

pnpm demo testet den HTTP-Handler, den Agent Loop und den SSE-Stream mit einem geskripteten Model. Außerdem ruft es die runSweep-Berechnung des Workers auf. Der Worker-Prozess und seine Database Queue werden nicht gestartet. Für die Demo benötigst du weder einen API-Key noch eine Datenbank oder Docker. pnpm check führt den Type-Checker, den Linter, den Formatter-Check und die Tests aus.

So startest du die echte API und den Worker:

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

Das sind die Commands, die ich im weiteren Verlauf dieses Artikels verwende. Das Repository kapselt die Low-Level-Node- und TypeScript-Commands hinter benannten pnpm-Scripts – ähnlich wie ein Python-Projekt uv run-Commands hinter make-Targets legen könnte. Füge npm install nicht in dieses pnpm-Repository ein. Verwende pnpm install, damit pnpm-lock.yaml die einzige Lockfile bleibt.

Was Node, npm, pnpm, TypeScript und tsx tun

Die ähnlichen Namen verbergen unterschiedliche Aufgaben:

• JavaScript ist die Sprache.
• Node.js ist die Runtime, ungefähr CPython für JavaScript. Installiere für dieses Projekt Node 24.
• Die npm Registry ist der Package-Index, ungefähr PyPI. Der npm-Command wird mit Node installiert und kann Packages aus dieser Registry installieren.
• pnpm ist der Package Manager, für den sich dieses Repository entschieden hat. Es installiert aus der npm Registry, verwaltet den Monorepo-Workspace und führt die in package.json deklarierten Commands aus.
• package.json ist das Projekt-Manifest und entspricht am ehesten pyproject.toml. Sein Abschnitt scripts gibt längeren Commands Namen wie demo, check und dev:api.
• TypeScript ist JavaScript mit statischen Typen. Sein tsc-Command prüft diese Typen. Das Repository führt ihn über pnpm check oder pnpm typecheck aus.
• tsx führt .ts-Dateien ohne separaten Build aus. Die Development-Scripts verwenden seinen watch-Mode, um die API oder den Worker nach einer Quellcodeänderung neu zu starten. Du rufst ihn in diesem Leitfaden nicht direkt auf.

Verwende in diesem Repository die pnpm-Scripts. Node ist die Runtime innerhalb dieser Scripts, und die bereitgestellte Dockerfile übernimmt die Production.

Der von Python abgeleitete Stack

Der Großteil dieser Zuordnung ist unspektakulär – und genau das ist die gute Nachricht. Drei Zeilen sind es nicht:

Das Begleit-Repository verwendet vier Libraries, die eine Erklärung wert sind.

Hono für die HTTP-Schicht

Express und Fastify sind Node-fokussierte Alternativen. Hono verwendet die Web-Standard-Request- und Response-APIs und bietet Adapter für Node und Serverless-Runtimes. Diese Portabilität ist für diese kleine Streaming-API nützlich, daher habe ich Hono gewählt.

Drizzle für SQL

Drizzle hält das Schema in TypeScript und benötigt keinen Client-Generation-Schritt. Außerdem stellt es Raw SQL bereit, wenn der Query Builder eine Postgres-Clause nicht sauber ausdrücken kann. Prisma würde ich wählen, wenn dessen generierter Client und das umgebende Tooling besser zum Team passen.

Biome für Linting und Formatting

Biome übernimmt Linting, Formatting und Import-Sortierung mit einer Binary und einer Konfigurationsdatei. Verwende weiterhin ESLint, wenn das Projekt von Custom Rules abhängt, die Biome nicht anbietet.

Vitest für Tests

Vitest führt die .ts-Tests des Begleit-Repositorys ohne separate Transform-Konfiguration aus.

Die hier verwendete TypeScript-Syntax lesen

Halte diese Tabelle bei den Service-Beispielen als Referenz bereit.

Verwende const, sofern sich die Binding nicht ändern muss. Verwende let für einen Counter, Accumulator oder eine andere Binding, die du neu zuweisen wirst.

Eine kurze Übersetzung von Enums

Für String-basierte States verwendet dieses Repository ein Object plus einen daraus abgeleiteten String-Union-Type:

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

Das Object stellt Status.Queued während der Runtime bereit. Die Zeile type erlaubt beim Type Checking nur "queued" oder "running". Zusammen erfüllen sie die beiden Rollen dieser Python-Deklaration:

from enum import Enum

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

Du musst lediglich das Muster erkennen. as const sorgt dafür, dass die Werte des Objects als exakte Strings erhalten bleiben, statt zu einem beliebigen string erweitert zu werden.

Die sieben semantischen Unterschiede, die Zeit kosten

Die Syntax-Tabelle bringt dich durch die Beispiele. Diese semantischen Unterschiede sind die Stellen, an denen Python-Gewohnheiten Bugs verursachen.

1. Leere Arrays und Objects sind truthy

Pythons Gewohnheit „leerer Container ist falsy“ überträgt sich am schlechtesten. if (results) ist auch bei einem leeren Array true. Schreibe if (results.length).

2. null und undefined sind unterschiedlich

null kennzeichnet normalerweise eine bewusst fehlende Value. undefined bedeutet normalerweise, dass eine Value fehlt oder noch nicht zugewiesen wurde, obwohl der Code sie explizit setzen kann. Library-Code gibt ständig undefined zurück. Der Unterschied wird relevant, wenn du einen Default schreibst. || verwendet die rechte Seite immer dann, wenn die linke Seite falsy ist. Dazu gehören 0, "" und false. ?? verwendet die rechte Seite dagegen nur für null und undefined. Daher ist 0 || 10 gleich 10, während 0 ?? 10 gleich 0 ist. So wird eine Batch-Größe von null stillschweigend zu zehn.

3. Ein Catch-Block erhält unknown

Es gibt kein except ValueError:. Ein einzelner catch-Block erhält alles. Da JavaScript das Werfen eines Strings, einer Zahl oder von null erlaubt, typisiert TypeScript den gefangenen Wert als unknown – den Typ „könnte buchstäblich alles sein“ unter strict. Das Begleit-Repository aktiviert strict, was neue Projekte grundsätzlich tun sollten. Um den Error zu untersuchen, musst du den Wert zunächst eingrenzen:

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

4. Promises starten sofort

Der Aufruf einer async-Funktion beginnt mit der Ausführung ihres Bodys und gibt ein Promise zurück. Ein Python-Coroutine-Object tut nichts, bis du es awaitest oder schedulst. Promise.all entspricht ungefähr asyncio.gather. Promise.allSettled entspricht ungefähr gather(..., return_exceptions=True), außer dass jedes Ergebnis als { status, value } oder { status, reason } verpackt wird.

Node verwaltet die Runtime-Scheduling-Logik. Der Prozess bleibt aktiv, solange aktive Handles oder Requests wie Timer und Sockets existieren. Ein gewöhnliches ausstehendes Promise allein hält Node nicht am Leben. Du umschließt das Programm nicht mit asyncio.run. In einem ES Module kannst du auf Top-Level await verwenden, wenn der Startup auf eine asynchrone Operation warten muss.

5. JavaScript besitzt einen gewöhnlichen numerischen Typ

Der JavaScript-number-Typ speichert Werte als 64-Bit-Gleitkommazahlen, ungefähr wie Pythons float. Der technische Standard für dieses Format heißt IEEE 754. Dezimalwerte sind Näherungen. Daher ist 0.1 + 0.2 nicht exakt 0.3, und Integer bleiben nur bis 2**53 - 1 oder 9,007,199,254,740,991 exakt.

Halte 64-Bit-IDs an Service-Grenzen als Strings. Die Konvertierung eines Postgres-bigint in einen JavaScript-number kann zu Rundungsfehlern führen. Für größere exakte Integer stellt JavaScript den separaten BigInt-Typ bereit, der sich nicht mit gewöhnlichen Zahlen mischen lässt.

6. Verwende Map, wenn du ein Dictionary im Python-Stil benötigst

In JavaScript erzeugt {} ein Object. Objects repräsentieren normalerweise Records mit benannten Feldern:

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

Ein Object ist keine saubere Key-Value-Tabelle wie ein Python-dict. Es erbt einige Namen aus JavaScript selbst. Das kann zu einem überraschenden Ergebnis führen:

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

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

Wenn ein externer String ein Feld aus einem Object auswählt, rufe vor dem Lesen Object.hasOwn auf. Für ein allgemeines Dictionary verwendest du Map. Map entspricht eher Pythons dict: Ein Key existiert nur, wenn dein Code ihn hinzufügt.

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

7. Extension bei relativen Imports angeben

Eine JavaScript-Quelldatei, die Code mit anderen Dateien teilt, heißt Module. Dieses Projekt verwendet das moderne Module-Format, ES Modules, meist als ESM abgekürzt. ES steht für ECMAScript, den formalen Namen der JavaScript-Sprache. In der Praxis ist ESM die im gesamten Projekt verwendete Syntax für import und export.

Bei einem relativen Import benötigt Node den exakten Dateinamen. Es errät nicht, ob ./env ./env.ts oder ./env.js bedeutet:

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

Imports aus installierten oder Workspace-Packages verwenden weiterhin den Package-Namen ohne Dateiendung:

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

Zod ist pydantic mit umgekehrtem Pfeil

In pydantic deklarierst du eine Klasse und erhältst einen Validator. In Zod deklarierst du einen Validator und leitest daraus den Typ ab. Dieselbe Single Source of Truth, aber in die entgegengesetzte Richtung.

Aus 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> liest einen statischen Typ aus dem Runtime-Schema aus. z.coerce.number() berücksichtigt, dass jeder definierte Wert in process.env (Node’s os.environ) ein String ist. Es erfüllt dieselbe Rolle wie die numerische Settings-Coercion von pydantic, auch wenn sich die jeweils akzeptierten exakten Strings unterscheiden. Dieses Muster folgt dem Prinzip pydantic-settings und wird einmal beim Startup ausgeführt. Eine ungültige Environment erzeugt dann einen verständlichen Startup-Error statt eines TypeError innerhalb eines Handlers.

Ein nacktes z.url() akzeptiert localhost:8000. Der URL-Standard behandelt alles vor dem ersten Doppelpunkt als Scheme. Daher interpretiert er localhost: als Protokoll namens „localhost“ und akzeptiert den String. Die Value erreicht anschließend den HTTP-Client und schlägt mit weniger Kontext fehl. Schema-Validation verschiebt Fehler nach vorn, erzwingt aber ein permissives Schema, wenn du genau dieses geschrieben hast.

Zod 4 liefert außerdem z.toJSONSchema mit. Daher benötigt dieses Projekt nicht die in älteren Tutorials verbreitete Dependency zod-to-json-schema. Das wird relevant, sobald ein Schema drei Consumer versorgen muss – genau darum geht es im Abschnitt „Ein Tool, drei Consumer“.

Der Service

Der Demo-Service dimensioniert LLM-Deployments. Ein Tool schlägt die Architekturkonstanten eines Models nach. Das andere schätzt dessen KV-cache-Footprint: den GPU-Speicher, der für Attention-Keys und -Values laufender Requests benötigt wird. Beide Tools führen bewusst nur triviale Arithmetik aus. Sie benötigen kein Netzwerk und liefern jedes Mal dasselbe Ergebnis. Dadurch lässt sich der Service ohne API-Key testen. Der KV-cache-Schätzer wird außerdem über das Model Context Protocol (MCP) veröffentlicht, sodass andere AI-Clients ihn aufrufen können.

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 ist die Datei, die den Workspace deklariert. Interne Packages erhalten einen Scoped-Namen wie @agent/core, wobei das Präfix @agent/ eine Naming Convention und kein Sprachfeature ist. Jedes Package deklariert seinen öffentlichen Einstiegspunkt in package.json. Diese Package-Grenze hängt nicht davon ab, welcher Command die Anwendung startet.

Dieser private Workspace verweist auf .ts-Quellcode, weil jeder Consumer Teil desselben Repositorys ist. Öffentliche npm-Packages veröffentlichen normalerweise JavaScript zusammen mit .d.ts-Typdeklarationen, damit gewöhnliche Node-Consumer weder den TypeScript-Runner noch das Build-Setup des Package-Autors benötigen.

Den Tool Loop einmal von Hand schreiben

Tool-Calling-Agent-Frameworks kapseln denselben grundlegenden Loop:

1. Rufe das Model mit Tool-Definitionen auf.
2. Validiere die angeforderten Tools und führe sie aus.
3. Hänge die Ergebnisse an die Messages an.
4. Rufe das Model erneut auf.

Schreibe diesen Loop einmal selbst. Dann wird das Verhalten des Frameworks zu einer Engineering-Entscheidung, die du begründen kannst.

Der Loop ist ein async function*, also ein Async Generator – exakt die Form von Pythons async def mit yield. Die HTTP-Route iteriert über diesen Generator, wandelt jedes Event in einen Server-Sent-Events-Frame um und sammelt den Text. Nach dem Ende des Streams ruft die Route storage.createRun einmal mit dem finalen Text auf. Tests rufen den Loop separat auf und sammeln seine Events in einem Array. Ein SSE-Frame ist ein Chunk einer langlebigen HTTP-Response. Der folgende Abschnitt erklärt das Format.

Der Loop hat zwei normale Enden. Eine Response ohne Tool Calls liefert done mit stopReason: "stop". Wenn das Model bis zu maxSteps weiterhin Tools anfordert, wird der Generator trotzdem abgeschlossen und liefert done mit stopReason: "max_steps". Behandle dieses Ergebnis als abgeschlossen, aber abgeschnitten: Der gesammelte Text kann unvollständig sein.

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

Die lokale Variable text enthält einen Model-Schritt. Der Loop verwendet sie in der Assistant-Message für den nächsten Schritt oder im finalen done-Event. Sie ist nicht der Accumulator auf Route-Ebene, der später persistiert wird.

Die Map partial ist der Teil, den Frameworks verbergen. Das SDK stellt String-Fragmente der Function Arguments bereit, die das serialisierte JSON an beliebigen Positionen teilen können. Mehrere parallele Calls können sich außerdem überlappen. Das Repository enthält einen Test, der {"model":"llama-3.1-8b",...} über vier Chunks verteilt.

Der zweite Teil, den du selbst schreiben solltest, ist das Verhalten bei fehlgeschlagener Validation:

// `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,
    };
}

Vor der Ausführung kann der Lookup das Tool verfehlen, JSON.parse kann die Arguments ablehnen oder Zod kann deren Shape zurückweisen. Jeder Fehler wird zu einer Message, die das Model liest. Während der Ausführung wird auch ein erwarteter ToolError zu einem Tool Result, damit das Model seinen Call korrigieren kann. Eine unerwartete Exception propagiert dagegen in den HTTP-Error-Pfad, statt als Domain Failure präsentiert zu werden. z.prettifyError verwandelt Zods Issue Tree in eine Message, mit der das Model arbeiten kann, statt in einen Stack Trace.

strict: true in einer OpenAI Function Definition fordert den Provider auf, das Decoding auf das Schema zu beschränken. Das hat keinen Zusammenhang mit TypeScripts strict-tsconfig-Flag. Es ähnelt vLLMs Guided Decoding, auch wenn sich unterstützte Schemas und Details der Durchsetzung unterscheiden. Dadurch entfällt ein Failure Mode, aber ein Self-Hosted Endpoint kann das Flag ignorieren. Die Arguments müssen außerdem JSON.parse überstehen.

Der Loop ruft /chat/completions auf, weil das Begleit-Repository OpenAI-kompatible Server anvisiert. vLLM, SGLang und Ollama dokumentieren diesen Endpoint, sodass OPENAI_BASE_URL denselben Client auf jeden dieser Server richten kann. Ihre Abdeckung der Responses API unterscheidet sich und ändert sich mit den Releases. Wenn du beide Seiten kontrollierst, prüfe vor der Entscheidung zwischen den beiden APIs die aktuelle Compatibility Page des Servers.

Dann zum AI SDK wechseln – und wissen, was du dafür eintauschst

Für spätere Projekte würde ich das Vercel AI SDK verwenden. Das Begleit-Repository implementiert denselben Agent zweimal, damit der Trade-off sichtbar wird. Beide Versionen emittieren denselben AgentEvent-Stream, sodass die HTTP-Schicht keinen Unterschied erkennt.

Das Begleit-Repository pinnt AI SDK 7.0.42 in packages/agent-core/package.json. Die Framework-Implementierung befindet sich in 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;
        // ...
    }
}

Das SDK entfernt fünf Teile Application Code:

• den Fragment-Accumulator
• JSON.parse und dessen Error Path
• den Call, der Zod auf die geparsten Arguments anwendet
• Provider-spezifische Message Assembly
• den Step Counter

stopWhen akzeptiert mehrere Bedingungen, darunter ein Step Limit oder einen bestimmten Tool Call. Der for await-Loop ändert sich nicht, wenn sich die Stop Policy ändert.

Was du aufgibst, ist direkte Kontrolle über Validation Failures. Der handgeschriebene Loop entscheidet, was das Model nach einem abgelehnten Call sieht. Das SDK stellt die experimentelle Option experimental_repairToolCall für einen eigenen Repair Callback bereit. Das aktuelle Begleit-Repository setzt diese Option nicht und verlässt sich daher auf das standardmäßige Invalid-Call-Handling des SDKs.

Der Trade-off wirkt auch in die andere Richtung. In der SDK-Version ist ein Provider-Wechsel auf den Provider-Adapter begrenzt. Das passende Provider-Package, Credentials, Configuration und Integration Tests werden weiterhin benötigt. Im Raw Loop gehören Provider-spezifische Request- und Stream-Verarbeitung zu deinem Code, den du ändern musst.

Ich schreibe den Loop im ersten Projekt von Hand und verwende in späteren Projekten das SDK. Für diese Lektion zahlst du nur einmal. Die Alternative besteht darin, beim ersten Production-Ausfall die Internals eines Frameworks zum ersten Mal lesen zu müssen.

Streaming über HTTP: Hono und SSE

Hono-Routen lesen sich wie FastAPI-Routen. Die Ergänzung ist zValidator, das die Aufgabe übernimmt, die FastAPI bei den Type Annotations der Handler-Signatur automatisch erledigt. Das c im folgenden Handler ist Hono’s Request Context – das Object, das FastAPI auf deine Parameter verteilt. deps ist ein Container für Dependencies, mit denen die App konstruiert wird, statt sie direkt zu importieren. runAgent ist eine davon, und der Testing-Abschnitt zeigt, was dir das bringt.

Aus 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 = "";
        try {
            await withSpan(
                "agent.run",
                { "agent.max_steps": body.maxSteps },
                async () => {
                    for await (const event of deps.runAgent({
                        message: body.message,
                        maxSteps: body.maxSteps,
                    })) {
                        if (event.type === "text") text += event.delta;

                        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: "succeeded",
            input: { message: body.message },
            output: { text },
        });
    });
});

Die Route behandelt derzeit jeden normal abgeschlossenen Generator gleich. Sie persistiert status: "succeeded" nach jedem Generator, der das Ende erreicht – auch nach einem, dessen finales Event stopReason: "max_steps" enthält. Dadurch gilt das Ergebnis als abgeschlossen, aber abgeschnitten oder unvollständig, nicht als bereits persistierter Failure. Production Code sollte das done-Event prüfen und eine explizite Policy anwenden. Beispielsweise könnte er einen separaten Truncated Status oder einen Review-/Retry-Pfad verwenden, bevor er Erfolg meldet.

zValidator validiert den Body und gibt c.req.valid("json") den vom Schema erzeugten Typ. Lässt du es weg, ist der Body als any typisiert – TypeScripts Opt-out, bei dem jeder Property-Zugriff kompiliert wird und nichts geprüft wird. Damit deaktivierst du den Type-Safety-Vorteil des Schemas.

Diese Route verwendet Server-Sent Events statt WebSockets. Der Server hält eine HTTP-Response offen, während er event: <name>- und data: <json>-Frames schreibt, und schließt sie nach dem finalen Event. Der Traffic fließt vom Server zum Client, was zu diesem Agent Stream passt. Ein WebSocket würde bidirektionales Messaging und ein Protocol Upgrade hinzufügen, das diese Route nicht benötigt.

Ein Midstream-Failure verändert die HTTP-Error-Berichterstattung. Sobald der erste Frame mit Status 200 gesendet wurde, kann der Server diese Response nicht mehr durch eine 500 ersetzen. Der catch-Block loggt den gefangenen Error, sendet ein konstantes Error-Event an den Client und kehrt zurück. Diese Rückgabe ist wichtig: Nur ein erfolgreich abgeschlossener Stream erreicht storage.createRun. Ein Generator, der done emittiert, einschließlich max_steps, gilt als normal abgeschlossener Stream und erreicht diesen Write.

Ein Test deckt diesen Pfad ab. Ein Generator liefert ein Text-Delta und wirft anschließend einen Error. Die Response bleibt 200, und ihr letzter Frame ist ein error-Event mit der konstanten Message Agent run failed. Der Logger behält den gefangenen Error für die serverseitige Diagnose. Jeder Client, der nur den Statuscode prüft, meldet einen fehlgeschlagenen Run als Erfolg.

app.ts trifft zwei kleinere Entscheidungen, die eine Erklärung verdienen. Es behandelt /healthz als Liveness-Endpoint, weshalb diese Route Postgres bewusst nicht anfasst. Ein Liveness-Failure während eines Datenbankausfalls könnte jede Replica neu starten, ohne die Dependency zu reparieren. Füge einen separaten Readiness-Check hinzu, wenn der Orchestrator den Traffic zu einer Instanz stoppen muss, die Postgres nicht erreicht. Die Error Paths loggen den gefangenen Error, geben aber einen konstanten String zurück. error.message in einen Response Body zu spiegeln ist der Weg, auf dem Connection Strings im Browser anderer Personen landen.

Der Celery-ähnliche Teil – ohne Celery

Lange Jobs gehören nicht in einen Request Handler. Die API fügt eine Row ein und gibt 202 zurück. Ein Worker übernimmt die Row.

Hier gibt es weder Redis noch BullMQ. PostgreSQL dokumentiert SKIP LOCKED für mehrere Consumer einer Queue-ähnlichen Tabelle. Die Clause gibt diesem kleinen Service eine At-Least-Once-Queue in einer Tabelle. Sie ist transaktional mit deinen übrigen Writes, und in docker-compose.yml gibt es einen Service weniger.

Die Claim Query in packages/agent-core/src/db/storage.ts lautet:

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()` exists but is undocumented; SKIP LOCKED rides in its second
    // argument.
    .for("update", { skipLocked: true });

Die Row wird für die Transaction gelockt, und ein anderer Worker, der dieselbe Query ausführt, überspringt sie, statt zu blockieren. Zwei parallele Claims erhalten daher nicht dieselbe nicht-veraltete Row. Ein Integration Test startet über Promise.all zwei Claims gleichzeitig und stellt fest, dass sie unterschiedliche Rows zurückgeben. Die naive Variante, SELECT ... LIMIT 1 gefolgt von UPDATE, schlägt diesen Test fehl: Beide Transactions lesen dieselbe Row, bevor eine von ihnen schreibt, und starten daher denselben Job.

Das ist At-Least-Once Execution, nicht Exactly-Once Execution. Die vollständige Query übernimmt außerdem eine running-Row, wenn ihr Lock länger als fünf Minuten besteht; der Demo-Worker verlängert diese Lease nicht. Ein aktiver Job, der länger als fünf Minuten läuft, kann daher zweimal geclaimt werden. Mache Jobs idempotent. Für lang laufende Arbeit fügst du entweder einen Lease Heartbeat hinzu oder setzt den Stale-Lock-Schwellenwert über die maximale Laufzeit.

Füge BullMQ hinzu, wenn du verzögerte Jobs, wiederholbare Schedules, Prioritäten, Rate Limits oder ein Dashboard benötigst. In Python würde ich denselben Schritt von einer Database Table zu Celery machen. Davor ist Redis ein weiterer Service, den du betreiben, überwachen und der On-Call-Person erklären musst.

Der Worker validiert erneut, was er aus jsonb liest:

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

Die Test-Suite übergibt dem Worker außerdem eine Row, deren seqLen ein String ist. Der Worker schlägt den Run fehl und pollt weiter, statt abzustürzen und dieselbe Poison Row endlos zu retrien.

CPU-Arbeit legt eine weitere Node-Beschränkung offen. Ein synchroner Callback läuft auf dem Event-Loop-Thread und wird nicht preempted. Eine for-Schleife, die zwei Sekunden lang Arithmetik ausführt, blockiert jeden Request, Timer und Liveness-Check dieses Prozesses für zwei Sekunden. Eine enge Schleife innerhalb eines async def blockiert asyncio auf dieselbe Weise. Beide Runtimes verlangen, dass du CPU-Arbeit explizit auslagerst.

await setTimeout(0) aus node:timers/promises (das Präfix node: bedeutet Standard Library, daher entspricht node:timers bei Node dem os bei Python) ist await asyncio.sleep(0). Der Sweep gibt nach jeder Batch-Größe die Kontrolle zurück, damit der Worker-Prozess Timer und andere Callbacks bedienen kann. Yielding macht CPU-Arbeit nicht parallel. Node worker_threads kann JavaScript parallel ausführen. Für reine Python-CPU-Arbeit unter einem gewöhnlichen GIL-aktivierten CPython-Build verwendest du statt eines Thread Pools einen Process Pool. Dieser Service verwendet keines von beiden. Halte umfangreiche numerische Berechnungen in Python, wo die unterstützenden Libraries bereits vorhanden sind, und verlagere sie vom API-Event-Loop.

Ein Tool, drei Consumer

EstimateKvCacheInput hat drei Consumer:

• Der handgeschriebene Loop konvertiert es mit z.toJSONSchema.
• Das AI SDK erhält es unverändert.
• Der MCP-Server veröffentlicht seine Shape.

Diese Wiederverwendung ist der Grund für packages/schemas.

Aus 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());

Vor dem Verbinden eines Clients sind zwei Details wichtig. Erstens verwendet ein auf diese Weise gestarteter Server stdin und stdout selbst zur Kommunikation mit dem Client. Jede Zeile ist eine JSON-RPC-Message. Ein zusätzlich ausgegebenes console.log, das JavaScript-Äquivalent zu print, beschädigt dann eine Message. Der Client trennt die Verbindung mit einem Parse Error, der keine Datei nennt. Sende alle Diagnostics an stderr.

Zweitens sollte ein Domain Failure isError: true mit einer Message zurückgeben. Das aufrufende Model kann den Call dann korrigieren, genau wie nach ungültigen Tool Arguments im Agent Loop.

Du kannst den Server einmal mit printf und einer Pipe ansteuern. Das solltest du tun, bevor du einen echten Client darauf richtest:

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

Der Probe entspricht der im README des Begleit-Repositorys verwendeten Protocol-Version. Für einen echten Client solltest du das SDK verwenden, statt JSON-RPC-Messages von Hand zu pflegen.

Einen Agent ohne API-Key testen

Vitest übernimmt die Rolle von pytest, aber die Struktur ist anders. describe gruppiert verwandte Tests. it und test definieren jeweils einen Test Case. test.each entspricht ungefähr parametrize, beforeEach stellt Setup pro Test bereit, vi.fn() erzeugt eine Mock-Funktion, und describe.skipIf überspringt eine Gruppe bedingt.

Agent Tests hängen von einer Entscheidung ab: runAgent erhält einen OpenAI-Client als Parameter, statt ihn selbst zu konstruieren. Der Fake ist ein Object mit einer chat.completions.create-Methode, die ein geskriptetes Async Iterable zurückgibt:

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
        // launder it through `unknown` first. A lie to the compiler, confined
        // to one line in a test file, which is the only place it belongs.
    } as unknown as OpenAI;
}

Die Tests teilen einen JSON-Argument-String über mehrere Chunks auf und verarbeiten zwei Tool Calls in einer Response. Außerdem decken sie ungültige batchSize, fehlerhaftes JSON, unbekannte Tool-Namen und ein Model ab, das so lange Tools aufruft, bis maxSteps es stoppt. Die Testdatei läuft deutlich unter einer Sekunde – ohne Netzwerk und ohne Key.

Integration Tests gegen Postgres verwenden describe.skipIf(!process.env.DATABASE_URL). Dadurch funktioniert pnpm test bei einem frischen Clone ohne laufendes Postgres, während CI die Tests durch Setzen der Variable aktiviert. Das Repository enthält 40 Tests. 36 davon laufen ohne Postgres oder API-Key.

Strukturierte Events mit Pino loggen

Pino erfüllt dieselbe Aufgabe wie structlog: ein JSON-Object pro Zeile, Child Logger mit gebundenen Feldern und explizite Redaction. Das Begleit-Repository konfiguriert es in packages/observability/src/logger.ts:

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

Ohne Redaction kann log.info({ req }, "...") einen Authorization-Header in das Log-Backend kopieren.

Application Work mit manuellen Spans tracen

Das Begleit-Repository verwendet OpenTelemetry für drei Spans auf Application-Ebene: agent.run, agent.tool und worker.sweep. Automatische HTTP- oder Postgres-Instrumentation wird nicht installiert. startTracing() in packages/observability/src/tracing.ts erzeugt einen NodeSDK mit einem OTLP-Trace-Exporter. Wenn OTEL_EXPORTER_OTLP_ENDPOINT fehlt, bleibt Tracing deaktiviert.

Die eigentliche Arbeit wird mit withSpan() aus derselben Datei umschlossen:

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

JavaScript besitzt keine Context-Manager-Syntax im Python-Stil. Hier ist der Callback der Block, den ein Python Context Manager umschließen würde. Der vollständige Helper zeichnet außerdem Exceptions auf, setzt den Span-Status und wirft die Exception anschließend erneut.

Automatische HTTP- und Database-Spans sind ein separates Feature. Sie benötigen die passenden Instrumentation-Packages und eine Initialisierung, bevor die instrumentierten Module geladen werden. Füge das nur hinzu, wenn diese Spans nützlich sind, und folge für die exakten Package-Versionen deines Deployments dem OpenTelemetry Node SDK Setup.

Das Monorepo in Docker ausliefern

Verwende die bereitgestellte Dockerfile. Der Container startet die API mit dem tsx-Loader. Beim Deployment musst du keinen TypeScript-Runner auswählen oder aufrufen.

Der Build verwendet pnpm fetch, sodass Dependency-Downloads gecacht bleiben, bis sich die Lockfile ändert. Anschließend verwendet er pnpm deploy, um die API und ihre Production Dependencies in ein eigenständiges Verzeichnis zu kopieren. Die Runtime Stage läuft als Non-Root-User node, und ihr Exec-Form-CMD erlaubt es der API, SIGTERM direkt für einen Graceful Shutdown zu empfangen.

Ein Drei-Wochen-Pfad

Erfahrene Python-Engineers können Material zu Variablen und Schleifen überspringen. Diese Sequenz konzentriert sich auf die Teile, die sich von Python unterscheiden. Die Build-Spalte ist der Zweck jeder Zeile. Die Lektüre unterstützt ihn.

Beginne mit den kostenlosen Total-TypeScript-Tutorials. Bezahle erst dann für fortgeschrittenes Material, wenn du mit Library-Grade Generics und Conditional Types arbeitest. Überspringe jeden „Intro to JavaScript“-Kurs und alles, was nach React aussieht, sofern das Produkt es nicht erfordert.

Für einen Überblick über Production Conventions ist goldbergyoni/nodebestpractices eine umfangreiche, von der Community gepflegte Checkliste. Prüfe Ratschläge, die das Runtime-Verhalten oder die Security betreffen, anhand der aktuellen Node-Dokumentation.

Trade-offs

Numerik in Python belassen

Node eignet sich gut für Orchestration, HTTP Serving und Streaming. Dauerhafte CPU-bound Math blockiert seinen Main-Event-Loop-Thread. Belasse vLLM und Training Code in Python, sofern ein gemessener Workload die Verlagerung nicht rechtfertigt.

Jede Grenze zur Runtime validieren

Eine TypeScript-Annotation prüft keinen HTTP Body, keine Environment Variable, kein Model-generiertes Tool Argument und keine aus jsonb gelesene Row. Jede Grenze benötigt ein Runtime-Schema.

SDK-Churn isolieren

AI SDK 6 ersetzte Experimental_Agent durch ToolLoopAgent und benannte die Agent Setting system in instructions um (AI-SDK-6-Migrationsleitfaden). Das Begleit-Repository ruft streamText direkt in AI SDK 7 auf und stellt seinen eigenen AgentEvent-Stream bereit. Diese Grenze hält die HTTP-Route unverändert, wenn sich der SDK-Code ändert.

Den handgeschriebenen Loop überspringen, wenn die Deadline wichtiger ist

Wenn du verstehen möchtest, welches Verhalten das SDK übernimmt, schreibe den Loop einmal selbst. Wenn du zuerst ausliefern musst und keinen Grund hast, Validation Failures anzupassen, beginne mit dem SDK.

Die wichtigsten Erkenntnisse

1. Installiere Node 24 und pnpm. Verwende anschließend die Repository-Scripts: pnpm demo, pnpm dev:api, pnpm dev:worker und pnpm check. Die Scripts verbergen die Low-Level-Runtime- und Type-Checker-Commands.
2. Runtime-Validation ist strukturell notwendig. Zod ist der gewählte Validator dieses Projekts. Statische Typen prüfen weder HTTP Bodies noch Environment Variables, Model Output oder Database Rows. Mit Zod deklarierst du ein Runtime-Schema und leitest den TypeScript-Typ mit z.infer ab.
3. Der Stack lässt sich größtenteils sauber abbilden: pnpm für uv, Hono für FastAPI, Drizzle für SQLAlchemy, Vitest für pytest und Biome für Ruff. Drei Zeilen sind kein einfacher Austausch: Validation, Type Checking und Job Queue.
4. Schreibe einen Agent Loop von Hand, wenn du die verborgenen Pfade verstehen oder anpassen musst: Fragment-Accumulation, Validation und Tool-Error-Feedback.
5. Mache den Loop zu einem Async Generator. Die HTTP-Route konsumiert seine AgentEvent-Werte, emittiert SSE-Frames, sammelt den Text und persistiert ihn nach dem Stream. Tests konsumieren den Generator separat, ohne Netzwerk und API-Key.
6. Postgres kann eine At-Least-Once-Queue bereitstellen. Mache Handler idempotent und erneuere die Lease oder dimensioniere sie über der maximalen Laufzeit für lange Jobs. Füge BullMQ hinzu, wenn du Delays, Prioritäten oder Schedules benötigst.
7. Verwende die bereitgestellte Dockerfile für Production. Sie paketiert die ausgewählte App, lädt TypeScript mit tsx, läuft als Non-Root-User und leitet Shutdown-Signale an den API-Prozess weiter.

Referenzen

Demo-Repository

• slavadubrov/typescript-agent-service – das im gesamten Artikel verwendete Monorepo: Hono-API mit SSE, zwei Implementierungen des Agent Loops, Drizzle Storage, Worker, MCP-Server und 40 Tests

Runtime und Sprache

• TypeScript nativ in Node.js ausführen – Nodes begrenzter TypeScript-Support und seine node_modules-Einschränkung
• TypeScript-Compiler-Optionen – strict und die übrigen vom Begleit-Repository konfigurierten Checks
• MDN JavaScript Guide – die Sprachreferenz, die du geöffnet lassen solltest
• javascript.info – modernes JavaScript-Tutorial. Lies die Kapitel zu Modules und Promises.

Tooling

• pnpm und pnpm-Installation – Package Manager, Workspaces und Setup
• Biome – Linting, Formatting und Import-Sortierung in einer Binary
• Vitest – Test Runner ohne Transform-Konfiguration
• Total TypeScript – kostenlose Tutorials plus ein kostenpflichtiger Track für fortgeschrittene Typen

Libraries

• Zod – Schema-Validation und Type Inference. Version 4 enthält z.toJSONSchema.
• Hono – Web-Standard-HTTP-Framework
• Drizzle ORM – SQL-first-TypeScript-ORM mit drizzle-kit-Migrations
• PostgreSQL-SELECT-Dokumentation – die FOR UPDATE ... SKIP LOCKED-Locking-Clause
• BullMQ – Redis-backed Queue, wenn eine Database Table nicht ausreicht

AI und Agents

• Vercel AI SDK – streamText, tool, stopWhen und Provider-Adapter
• openai/openai-node – der offizielle TypeScript-Client
• MCP TypeScript SDK und MCP-Spezifikation – Server und Clients entwickeln

Conventions

• goldbergyoni/nodebestpractices – von der Community gepflegte Checkliste für Production Conventions