KI & LLM APIs 5. Mai 2026 11 min Lesezeit

OpenAI Assistants API mit Claude Code: KI-Agenten bauen 2026

Die OpenAI Assistants API ermöglicht persistente KI-Agenten mit Gedächtnis, Tool-Nutzung und Datei-Verarbeitung — ohne eigenes State-Management. Claude Code kennt die API vollständig und generiert produktionsreifen TypeScript-Code für Threads, Runs, File Search, Code Interpreter und Streaming.

Assistants API vs Chat Completions: Wann was nutzen?

Die Chat Completions API ist zustandslos — jeder Request ist eine neue Konversation. Die Assistants API dagegen verwaltet Threads (Gesprächsverläufe), Runs (Ausführungen) und Tools serverseitig. Das spart Entwicklungsaufwand und ermöglicht natürliche Multi-Turn-Dialoge ohne selbst den Kontext zu serialisieren.

Eigenschaft	Chat Completions	Assistants API
State-Management	✗ Manuell (du schickst History)	✓ Serverseitig (Threads)
Persistenz	✗ Kein eingebautes Gedächtnis	✓ Thread bleibt bestehen
File Search	✗ Manuelles RAG nötig	✓ Vector Store eingebaut
Code Interpreter	✗ Nicht verfügbar	✓ Python-Sandbox eingebaut
Kosten	Günstiger pro Token	Höher (Tool-Aufrufe extra)
Latenz	Geringer (synchron)	Höher (Run-Poll-Schleife)

Faustregel: Für einfache Q&A und einmalige Aufgaben → Chat Completions. Für persistente Agenten, Datei-Analyse und interaktive Chatbots mit Gedächtnis → Assistants API.

SetupInstallation und Initialisierung

// Prompt: "Richte OpenAI Assistants API in TypeScript ein"

import OpenAI from "openai";

// npm install openai
const openai = new OpenAI({
  apiKey: process.env.OPENAI_API_KEY, // Niemals hardcoden!
});

// Typen für typsichere Entwicklung:
import type {
  Assistant,
  Thread,
  Run,
  Message,
} from "openai/resources/beta/assistants";

// Basis-Konfiguration prüfen:
async function verifySetup(): Promise<void> {
  const models = await openai.models.list();
  console.log(`✓ API-Key gültig, ${models.data.length} Modelle verfügbar`);
}

Assistant und Thread erstellen

Ein Assistant ist die Konfiguration deines KI-Agenten — Persönlichkeit, Fähigkeiten (Tools) und Modell. Ein Thread ist der Gesprächsverlauf. Claude Code generiert beide mit sauberer Trennung: Assistant einmal erstellen, Thread pro Nutzer-Session.

ThreadAssistant erstellen und wiederverwenden

// Prompt: "Erstelle einen persistenten KI-Assistenten mit TypeScript"

// EINMALIG: Assistant erstellen (ID in DB/Env speichern!)
async function createAssistant(): Promise<Assistant> {
  const assistant = await openai.beta.assistants.create({
    name: "Produktionsberater",
    instructions: `Du bist ein freundlicher Produktberater für unser SaaS-Tool.
      Beantworte Fragen präzise auf Deutsch.
      Nutze die verfügbaren Dokumente für genaue Antworten.
      Bei Unklarheiten frage gezielt nach.`,
    model: "gpt-4o",
    tools: [
      { type: "file_search" },
      { type: "code_interpreter" },
    ],
  });

  console.log(`Assistant erstellt: ${assistant.id}`);
  // Speichern: process.env.ASSISTANT_ID = assistant.id
  return assistant;
}

// PRO NUTZER-SESSION: Thread erstellen
async function createThread(): Promise<Thread> {
  const thread = await openai.beta.threads.create({
    metadata: {
      userId: "user-123",
      sessionStart: new Date().toISOString(),
    },
  });

  console.log(`Thread erstellt: ${thread.id}`);
  return thread;
}

// Thread wiederverwenden — Verlauf bleibt erhalten:
async function sendMessage(
  threadId: string,
  content: string
): Promise<string> {
  // Nachricht in Thread einfügen:
  await openai.beta.threads.messages.create(threadId, {
    role: "user",
    content,
  });

  // Run starten (Assistant verarbeitet Thread):
  const run = await openai.beta.threads.runs.createAndPoll(threadId, {
    assistant_id: process.env.ASSISTANT_ID!,
  });

  if (run.status !== "completed") {
    throw new Error(`Run fehlgeschlagen: ${run.status}`);
  }

  // Letzte Antwort abrufen:
  const messages = await openai.beta.threads.messages.list(threadId, {
    order: "desc",
    limit: 1,
  });

  const lastMsg = messages.data[0];
  if (lastMsg.content[0].type === "text") {
    return lastMsg.content[0].text.value;
  }
  return "";
}

Best Practice: Assistant-ID einmalig erstellen und in der Umgebungsvariablen ASSISTANT_ID speichern. Thread-ID pro Nutzer in der Datenbank persistieren — so läuft der Gesprächsverlauf über mehrere Sessions weiter.

ThreadThread-Verlauf laden

// Alle Nachrichten eines Threads abrufen:
async function getThreadHistory(
  threadId: string
): Promise<{role: string; content: string}[]> {
  const messages = await openai.beta.threads.messages.list(threadId, {
    order: "asc", // Chronologisch
    limit: 100,
  });

  return messages.data.map((msg) => ({
    role: msg.role,
    content: msg.content
      .filter((c) => c.type === "text")
      .map((c) => (c as any).text.value)
      .join("\n"),
  }));
}

// Thread löschen wenn nicht mehr benötigt:
await openai.beta.threads.del(threadId);

File Search: Semantische Suche über Dokumente

File Search ist eingebautes RAG (Retrieval Augmented Generation) ohne eigene Vektordatenbank. Du lädst Dateien hoch, erstellst einen Vector Store, und der Assistant durchsucht automatisch relevante Passagen — Claude Code generiert den kompletten Upload-Workflow.

File SearchVector Store erstellen und Dateien hochladen

// Prompt: "Füge der Assistants API File Search für unsere Docs hinzu"

import fs from "fs";
import path from "path";

// 1. Vector Store erstellen:
async function createVectorStore(name: string) {
  const vectorStore = await openai.beta.vectorStores.create({
    name,
    expires_after: {
      anchor: "last_active_at",
      days: 30, // Automatisch löschen nach 30 Tagen Inaktivität
    },
  });

  console.log(`Vector Store erstellt: ${vectorStore.id}`);
  return vectorStore;
}

// 2. Dateien hochladen und in Vector Store einfügen:
async function uploadDocsToVectorStore(
  vectorStoreId: string,
  docsDir: string
): Promise<void> {
  const files = fs
    .readdirSync(docsDir)
    .filter((f) => f.endsWith(".pdf") || f.endsWith(".md"));

  // Batch-Upload für Effizienz:
  const fileStreams = files.map((f) =>
    fs.createReadStream(path.join(docsDir, f))
  );

  const batch = await openai.beta.vectorStores.fileBatches.uploadAndPoll(
    vectorStoreId,
    { files: fileStreams }
  );

  console.log(`${batch.file_counts.completed} Dateien indexiert`);
  console.log(`${batch.file_counts.failed} Dateien fehlgeschlagen`);
}

// 3. Vector Store mit Assistant verknüpfen:
await openai.beta.assistants.update(process.env.ASSISTANT_ID!, {
  tool_resources: {
    file_search: {
      vector_store_ids: [vectorStoreId],
    },
  },
});

// 4. Oder: Vector Store direkt im Thread-Scope (pro Nutzer):
const thread = await openai.beta.threads.create({
  tool_resources: {
    file_search: {
      vector_store_ids: [vectorStoreId],
    },
  },
});

Unterstützte Formate: PDF, DOCX, TXT, MD, HTML, PPTX, JSON und mehr. Pro Datei max. 512 MB. Vector Store-Suche funktioniert auch über mehrere Stores gleichzeitig — ideal für mandantenfähige Anwendungen.

File SearchZitierungen aus Suchergebnissen extrahieren

// File Search liefert automatisch Quellenangaben:
async function askWithCitations(
  threadId: string,
  question: string
): Promise<void> {
  await openai.beta.threads.messages.create(threadId, {
    role: "user",
    content: question,
  });

  const run = await openai.beta.threads.runs.createAndPoll(threadId, {
    assistant_id: process.env.ASSISTANT_ID!,
  });

  const messages = await openai.beta.threads.messages.list(threadId, {
    order: "desc",
    limit: 1,
    run_id: run.id,
  });

  const msg = messages.data[0];
  for (const block of msg.content) {
    if (block.type !== "text") continue;

    // Annotations enthalten Quellenverweise:
    let text = block.text.value;
    for (const annotation of block.text.annotations) {
      if (annotation.type === "file_citation") {
        // Dateiname abrufen:
        const file = await openai.files.retrieve(
          annotation.file_citation.file_id
        );
        text = text.replace(
          annotation.text,
          ` [Quelle: ${file.filename}]`
        );
      }
    }
    console.log(text);
  }
}

Code Interpreter: Python-Code in der KI-Sandbox

Der Code Interpreter führt Python-Code in einer isolierten Sandbox aus — ohne eigenen Server. Das Modell schreibt Code, führt ihn aus, sieht das Ergebnis und kann iterieren. Claude Code nutzt dieses Muster für Datenanalyse, Diagramme und Datei-Generierung.

Code InterpreterDatenanalyse und Diagramme

// Prompt: "Baue einen Datenanalyse-Assistenten mit Code Interpreter"

async function analyzeData(
  threadId: string,
  csvFilePath: string,
  analysisRequest: string
): Promise<string[]> {
  // CSV-Datei hochladen:
  const fileStream = fs.createReadStream(csvFilePath);
  const uploadedFile = await openai.files.create({
    file: fileStream,
    purpose: "assistants",
  });

  // Analyse-Anfrage mit Datei-Referenz:
  await openai.beta.threads.messages.create(threadId, {
    role: "user",
    content: [
      {
        type: "text",
        text: `Analysiere diese Datei: ${analysisRequest}
          Erstelle auch ein übersichtliches Diagramm.`,
      },
      {
        type: "image_file",
        image_file: { file_id: uploadedFile.id },
      },
    ],
    attachments: [
      {
        file_id: uploadedFile.id,
        tools: [{ type: "code_interpreter" }],
      },
    ],
  });

  const run = await openai.beta.threads.runs.createAndPoll(threadId, {
    assistant_id: process.env.ASSISTANT_ID!,
  });

  // Generierte Bilder (Diagramme) abrufen:
  const messages = await openai.beta.threads.messages.list(threadId, {
    order: "desc",
    run_id: run.id,
  });

  const imageFileIds: string[] = [];
  for (const msg of messages.data) {
    for (const block of msg.content) {
      if (block.type === "image_file") {
        imageFileIds.push(block.image_file.file_id);
      }
    }
  }

  return imageFileIds; // IDs für weiteren Download
}

// Generierte Datei herunterladen:
async function downloadGeneratedFile(
  fileId: string,
  outputPath: string
): Promise<void> {
  const response = await openai.files.content(fileId);
  const buffer = Buffer.from(await response.arrayBuffer());
  fs.writeFileSync(outputPath, buffer);
  console.log(`Datei gespeichert: ${outputPath}`);
}

Python-Libraries verfügbar: pandas, numpy, matplotlib, seaborn, scipy, sklearn — alle vorinstalliert in der Sandbox. Das Modell kann iterativ debuggen wenn der erste Versuch fehlschlägt.

Code InterpreterExcel- und PDF-Generierung

// Code Interpreter kann komplexe Dateien generieren:
async function generateReport(
  threadId: string,
  data: Record<string, number[]>
): Promise<string> {
  const dataJson = JSON.stringify(data, null, 2);

  await openai.beta.threads.messages.create(threadId, {
    role: "user",
    content: `Erstelle aus diesen Daten eine Excel-Datei mit:
      1. Übersichtstabelle auf Sheet 1
      2. Diagramm auf Sheet 2
      3. Statistiken (Mittelwert, Median, Standardabweichung)

      Daten: ${dataJson}`,
  });

  const run = await openai.beta.threads.runs.createAndPoll(threadId, {
    assistant_id: process.env.ASSISTANT_ID!,
  });

  // Excel-Datei aus Antwort extrahieren:
  const messages = await openai.beta.threads.messages.list(
    threadId, { run_id: run.id, order: "desc", limit: 1 }
  );

  for (const block of messages.data[0]?.content ?? []) {
    if (block.type === "text") {
      // File-IDs in Annotations suchen:
      for (const ann of block.text.annotations) {
        if (ann.type === "file_path") {
          return ann.file_path.file_id;
        }
      }
    }
  }
  throw new Error("Keine Datei generiert");
}

Function Calling: Externe APIs und Aktionen

Mit Function Calling definierst du Tools die der Assistant aufrufen darf — CRM-Updates, Datenbank-Abfragen, externe APIs. Der Run-Status-Loop überwacht wann Eingaben benötigt werden. Claude Code generiert den kompletten Loop mit Fehlerbehandlung.

Function CallingTools definieren und Run-Status-Loop

// Prompt: "Implementiere Function Calling für CRM-Integration"

// 1. Tools beim Assistant registrieren:
const crmAssistant = await openai.beta.assistants.update(
  process.env.ASSISTANT_ID!,
  {
    tools: [
      {
        type: "function",
        function: {
          name: "get_customer",
          description: "Ruft Kundendaten anhand der E-Mail-Adresse ab",
          parameters: {
            type: "object",
            properties: {
              email: {
                type: "string",
                description: "E-Mail-Adresse des Kunden",
              },
            },
            required: ["email"],
          },
        },
      },
      {
        type: "function",
        function: {
          name: "update_ticket",
          description: "Aktualisiert ein Support-Ticket",
          parameters: {
            type: "object",
            properties: {
              ticket_id: { type: "string" },
              status: {
                type: "string",
                enum: ["open", "in_progress", "resolved", "closed"],
              },
              note: { type: "string", description: "Interne Notiz" },
            },
            required: ["ticket_id", "status"],
          },
        },
      },
    ],
  }
);

// 2. Run-Status-Loop mit Tool-Output-Einreichung:
async function runWithTools(
  threadId: string,
  userMessage: string
): Promise<string> {
  await openai.beta.threads.messages.create(threadId, {
    role: "user",
    content: userMessage,
  });

  let run = await openai.beta.threads.runs.create(threadId, {
    assistant_id: process.env.ASSISTANT_ID!,
  });

  // Poll-Loop: Run beobachten bis abgeschlossen
  while (true) {
    run = await openai.beta.threads.runs.retrieve(threadId, run.id);

    if (run.status === "completed") break;

    if (run.status === "requires_action") {
      // Tool-Calls ausführen:
      const toolCalls =
        run.required_action!.submit_tool_outputs.tool_calls;

      const toolOutputs = await Promise.all(
        toolCalls.map(async (tc) => {
          const args = JSON.parse(tc.function.arguments);
          let output: string;

          switch (tc.function.name) {
            case "get_customer":
              output = JSON.stringify(await fetchCustomer(args.email));
              break;
            case "update_ticket":
              output = JSON.stringify(
                await updateTicket(args.ticket_id, args.status, args.note)
              );
              break;
            default:
              output = JSON.stringify({ error: "Unbekanntes Tool" });
          }

          return { tool_call_id: tc.id, output };
        })
      );

      // Tool-Outputs einreichen → Run läuft weiter:
      run = await openai.beta.threads.runs.submitToolOutputsAndPoll(
        threadId,
        run.id,
        { tool_outputs: toolOutputs }
      );
      continue;
    }

    if (["failed", "cancelled", "expired"].includes(run.status)) {
      throw new Error(`Run fehlgeschlagen: ${run.status} — ${run.last_error?.message}`);
    }

    // Kurze Pause um API-Rate-Limits zu schonen:
    await new Promise((resolve) => setTimeout(resolve, 500));
  }

  // Antwort abrufen:
  const msgs = await openai.beta.threads.messages.list(threadId, {
    order: "desc",
    limit: 1,
  });
  return (msgs.data[0].content[0] as any).text.value;
}

Achtung: Run-Status kann auch in_progress, queued oder cancelling sein. Implementiere immer einen Timeout (z.B. max. 60 Sekunden), um unendliche Loops zu vermeiden. submitToolOutputsAndPoll erledigt das Poll-Loop intern — nutze es wo möglich.

Streaming Runs: Echtzeit-Antworten

Statt auf das Ende eines Runs zu warten streamt die API Token für Token — entscheidend für gute UX. Mit stream: true empfängst du delta events die direkt in die UI gepusht werden. Claude Code generiert fertige React-Integration mit Server-Sent Events.

StreamingServer-Side Streaming mit Delta Events

// Prompt: "Implementiere Streaming für die Assistants API"

// Node.js / Express Backend:
import { AssistantStreamEvent } from "openai/resources/beta/assistants";
import { Stream } from "openai/streaming";

async function streamResponse(
  threadId: string,
  userMessage: string,
  onToken: (token: string) => void,
  onComplete: (fullText: string) => void
): Promise<void> {
  await openai.beta.threads.messages.create(threadId, {
    role: "user",
    content: userMessage,
  });

  const stream = openai.beta.threads.runs.stream(threadId, {
    assistant_id: process.env.ASSISTANT_ID!,
  });

  let fullText = "";

  // Event-Handler für verschiedene Delta-Typen:
  stream
    .on("textDelta", (delta) => {
      if (delta.value) {
        fullText += delta.value;
        onToken(delta.value); // Token direkt an UI senden
      }
    })
    .on("toolCallDelta", (delta) => {
      // Tool-Calls während Streaming (z.B. Code Interpreter)
      console.log("Tool wird ausgeführt...", delta.type);
    })
    .on("runStepDone", (step) => {
      console.log(`Step abgeschlossen: ${step.type}`);
    })
    .on("end", () => {
      onComplete(fullText);
    })
    .on("error", (error) => {
      console.error("Stream-Fehler:", error);
    });

  await stream.finalRun();
}

StreamingReact-Integration mit Server-Sent Events

// Backend: Express Route für SSE
app.post("/api/chat/stream", async (req, res) => {
  const { threadId, message } = req.body;

  // SSE-Header setzen:
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("Connection", "keep-alive");

  await openai.beta.threads.messages.create(threadId, {
    role: "user",
    content: message,
  });

  const stream = openai.beta.threads.runs.stream(threadId, {
    assistant_id: process.env.ASSISTANT_ID!,
  });

  stream.on("textDelta", (delta) => {
    if (delta.value) {
      res.write(`data: ${JSON.stringify({ token: delta.value })}\n\n`);
    }
  });

  stream.on("end", () => {
    res.write(`data: ${JSON.stringify({ done: true })}\n\n`);
    res.end();
  });

  await stream.finalRun();
});

// Frontend React-Hook für Streaming:
function useStreamingChat(threadId: string) {
  const [response, setResponse] = React.useState("");
  const [isStreaming, setIsStreaming] = React.useState(false);

  const sendMessage = React.useCallback(async (message: string) => {
    setResponse("");
    setIsStreaming(true);

    const eventSource = new EventSource(
      `/api/chat/stream?threadId=${threadId}`
    );

    // POST via fetch, dann SSE empfangen:
    await fetch("/api/chat/stream", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ threadId, message }),
    });

    // Einfachere Alternative: ReadableStream:
    const res = await fetch("/api/chat/stream", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ threadId, message }),
    });

    const reader = res.body!.getReader();
    const decoder = new TextDecoder();

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      const lines = decoder.decode(value).split("\n");
      for (const line of lines) {
        if (!line.startsWith("data: ")) continue;
        const data = JSON.parse(line.slice(6));
        if (data.done) { setIsStreaming(false); break; }
        if (data.token) setResponse((prev) => prev + data.token);
      }
    }
  }, [threadId]);

  return { response, isStreaming, sendMessage };
}

React-Komponente: Mit isStreaming kannst du einen blinkenden Cursor zeigen. Das Token-by-Token-Update via setResponse((prev) => prev + token) ist performant — React batcht die Updates automatisch in modernen Versionen.

StreamingStreaming mit Function Calling kombinieren

// Stream mit Tool-Calls — komplexestes Szenario:
async function streamWithTools(
  threadId: string,
  userMessage: string,
  onToken: (t: string) => void
): Promise<void> {
  await openai.beta.threads.messages.create(threadId, {
    role: "user",
    content: userMessage,
  });

  const runner = openai.beta.threads.runs.stream(threadId, {
    assistant_id: process.env.ASSISTANT_ID!,
  });

  for await (const event of runner) {
    switch (event.event) {
      case "thread.message.delta":
        for (const delta of event.data.delta.content ?? []) {
          if (delta.type === "text" && delta.text?.value) {
            onToken(delta.text.value);
          }
        }
        break;

      case "thread.run.requires_action": {
        // Tool-Calls auch während Streaming möglich:
        const toolCalls =
          event.data.required_action!.submit_tool_outputs.tool_calls;
        const outputs = await Promise.all(
          toolCalls.map(async (tc) => ({
            tool_call_id: tc.id,
            output: await executeToolCall(tc),
          }))
        );
        // Tool-Outputs streamed einreichen:
        runner.submitToolOutputsStream({
          tool_outputs: outputs,
        });
        break;
      }

      case "thread.run.completed":
        console.log("Run abgeschlossen");
        break;

      case "thread.run.failed":
        throw new Error(`Run fehlgeschlagen: ${event.data.last_error?.message}`);
    }
  }
}

Produktions-Checkliste: Assistants API

Best PracticesWas Claude Code immer implementiert

Assistant-ID cachen: Einmal erstellen, in ASSISTANT_ID Env speichern — nicht bei jedem Request neu erstellen
Thread-IDs persistieren: Pro Nutzer in Datenbank speichern für Verlauf über Sessions hinweg
Run-Timeout: Immer max. Wartezeit definieren (60s) — Runs können bei Tool-Problemen hängen
Error States: failed, cancelled, expired explizit behandeln mit Nutzer-Feedback
Retry-Logic: Bei 429 Rate Limit mit exponential backoff wiederholen
Vector Store Cleanup: Alte Stores und Files regelmäßig löschen (Kosten!)
Streaming bevorzugen: Bessere UX durch sofortiges Feedback statt Warten

Kosten-Hinweis: File Search und Code Interpreter erzeugen zusätzliche Kosten pro Nutzung (aktuell ~$0.20/1000 File Search-Calls bzw. ~$0.03/Session Code Interpreter). Vector Stores kosten $0.10/GB/Tag. Claude Code kennt diese Pricing-Details und kommentiert sie direkt im generierten Code.

KI-Agenten Modul im Kurs

Im Claude Code Mastery Kurs: vollständiges Assistants API Modul — Threads, Runs, File Search, Code Interpreter, Function Calling und Streaming. Mit fertigen TypeScript-Templates für produktionsreife KI-Agenten in 2026.

14 Tage kostenlos testen →