@ctx0/sdk

Drop-in LLM observability for TypeScript/Node.js. One import change, all calls logged automatically.

Install

npm install @ctx0/sdk

Usage

import { configure, OpenAI } from "@ctx0/sdk";

configure({ apiKey: "your-api-key" });

const client = new OpenAI({ apiKey: "sk-..." });
const response = await client.chat.completions.create({
  model: "gpt-4o",
  messages: [{ role: "user", content: "Hello" }],
});

Supports OpenAI, AzureOpenAI, and Anthropic.

You can also set the LLM_OBSERVATORY_API_KEY environment variable instead of passing apiKey to configure() — the SDK picks it up automatically.

Langfuse integration

Already using Langfuse? wrapLangfuse() captures all Langfuse traces, generations, and spans — including metadata like userId, sessionId, and tags — and forwards them to Context0 automatically. Zero code changes to your existing Langfuse setup.

Requires Langfuse TS SDK v4+ (@langfuse/tracing, @langfuse/otel).

Setup

import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";
import { configure, wrapLangfuse } from "@ctx0/sdk";

// 1. Configure Context0
configure({ apiKey: "your-api-key" });

// 2. Initialize Langfuse with OTel (standard Langfuse v4 setup)
const sdk = new NodeSDK({
  spanProcessors: [
    new LangfuseSpanProcessor({
      publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
      secretKey: process.env.LANGFUSE_SECRET_KEY!,
    }),
  ],
});
sdk.start();

// 3. Call once — adds Context0 processor to the TracerProvider
wrapLangfuse();

What gets captured

After calling wrapLangfuse(), every Langfuse observation is automatically forwarded to Context0:

import { startActiveObservation } from "@langfuse/tracing";
import OpenAI from "openai";

const client = new OpenAI({ apiKey: "sk-..." });

// Per-request metadata (userId, sessionId, tags) — captured automatically
const reply = await startActiveObservation(
  {
    name: "user-question",
    userId: "alice",
    sessionId: "sess-1",
    tags: ["production", "chat"],
    asType: "generation",
    model: "gpt-4o",
  },
  async (span) => {
    const messages = [{ role: "user" as const, content: "What is observability?" }];
    span.update({ input: messages });

    const response = await client.chat.completions.create({
      model: "gpt-4o",
      messages,
    });

    const content = response.choices[0].message.content ?? "";
    span.update({
      output: { role: "assistant", content },
      usageDetails: {
        input: response.usage?.prompt_tokens ?? 0,
        output: response.usage?.completion_tokens ?? 0,
      },
    });
    return content;
  },
);

// Multi-step traces (RAG pipeline) — captured automatically
await startActiveObservation(
  { name: "rag-pipeline", userId: "alice", sessionId: "sess-1" },
  async () => {
    // Nested spans create parent-child relationships in Context0
    const docs = await startActiveObservation(
      { name: "retrieval", asType: "span" },
      async (span) => {
        const results = await vectorDb.search(query);
        span.update({ input: query, output: results });
        return results;
      },
    );

    return await startActiveObservation(
      { name: "completion", asType: "generation", model: "gpt-4o" },
      async (span) => {
        // LLM call — input, output, usage all captured
        span.update({ input: messages });
        const response = await client.chat.completions.create({ model: "gpt-4o", messages });
        span.update({ output: response.choices[0].message, usageDetails: { ... } });
        return response.choices[0].message.content;
      },
    );
  },
);

Works with startActiveObservation(), observe(), and all Langfuse v4 patterns.

wrap() — per-client alternative

If you only want to observe a specific client instance (rather than all Langfuse observations globally), use wrap() instead:

import { observeOpenAI } from "langfuse";
import { configure, wrap } from "@ctx0/sdk";

configure({ apiKey: "your-api-key" });

const client = wrap(observeOpenAI(new OpenAI({ apiKey: "sk-..." })));

// Both Langfuse AND Context0 capture this call
const response = await client.chat.completions.create({
  model: "gpt-4o",
  messages: [{ role: "user", content: "Hello" }],
});

wrap() works with any client that has a .chat (OpenAI-shaped) or .messages (Anthropic-shaped) interface. If the client is already a Context0 wrapper, it's returned as-is — no double-wrapping.

When to use which: Use wrapLangfuse() to capture everything Langfuse sees globally (recommended). Use wrap() if you only want Context0 on specific client instances.

Tracing

Group related LLM calls into traces to see your full pipeline as a tree.

observe() wrapper

Automatically creates a trace (top-level) or span (nested). Captures function args as input and return value as output.

import { configure, OpenAI, observe } from "@ctx0/sdk";

const client = new OpenAI();

const retrieveDocs = observe(async function retrieveDocs(query: string) {
  return vectorDb.search(query);
});

const ragPipeline = observe(async function ragPipeline(question: string) {
  const docs = await retrieveDocs(question);  // child span
  const response = await client.chat.completions.create({  // auto-captured as generation
    model: "gpt-4o",
    messages: [{ role: "user", content: `${docs}\n\n${question}` }],
  });
  return response.choices[0].message.content;
});

Options:

• observe({ name: "custom-name" }, fn) — override the span name (default: function name)
• observe({ captureInput: false }, fn) — don't log args (for sensitive data)
• observe({ captureOutput: false }, fn) — don't log return value

Works with both sync and async functions.

Inline tracing

Use trace() and span() for inline code blocks that aren't standalone functions:

import { trace, span } from "@ctx0/sdk";

trace("rag-pipeline", { input: { query: q } }, (t) => {
  span("vector-search", (s) => {
    const results = search(q);
    s.setOutput(results);
  });

  const response = await client.chat.completions.create({
    model: "gpt-4o",
    messages: [...],
  });
  t.setOutput(response.choices[0].message.content);
});

Mix freely — observe() and span() compose within the same trace.

Cross-service propagation

Pass trace context across services (e.g., API to SQS worker) via W3C traceparent:

import { trace, getCurrentTraceparent, emitCompletedSpan } from "@ctx0/sdk";

// Producer — get traceparent inside a trace
trace("api-request", () => {
  const traceparent = getCurrentTraceparent();
  sqs.sendMessage({ Body: JSON.stringify({ traceparent, ... }) });
});

// Consumer — restore context
const msg = JSON.parse(sqsMessage.Body);
trace("worker", { traceparent: msg.traceparent }, () => {
  await client.chat.completions.create(...);  // appears as child in same trace
});

Use emitCompletedSpan() to backfill spans with explicit start/end times (e.g., queue wait duration).

Serverless / short-lived environments

In AWS Lambda, Convex, Vercel Functions, or any environment where the process freezes or terminates after your handler returns, buffered events may be lost before the background flush fires.

Option 1 — immediate mode (recommended): Send each event as it happens, no batching:

configure({
  apiKey: "your-api-key",
  flushMode: "immediate",
});

Option 2 — explicit flush: Keep the default batch mode but flush before returning:

import { configure, OpenAI, flush } from "@ctx0/sdk";

configure({ apiKey: "your-api-key" });
const client = new OpenAI({ apiKey: "sk-..." });

export async function handler(event: any) {
  const response = await client.chat.completions.create({
    model: "gpt-4o",
    messages: [{ role: "user", content: event.prompt }],
  });

  await flush(); // wait for all buffered events to send before the process freezes

  return { statusCode: 200, body: response.choices[0].message.content };
}

flushMode: "immediate" has slightly higher per-request overhead but guarantees no events are lost. Use it when you can't control when the process shuts down.

Using with Langfuse in serverless

If you're using wrapLangfuse(), Langfuse also buffers spans and needs flushing. The OTel SDK's shutdown() flushes all registered span processors — both Langfuse and Context0 — in one call:

import { NodeSDK } from "@opentelemetry/sdk-node";

// At the end of your handler:
await sdk.shutdown(); // flushes both Langfuse and Context0

If you're using flushMode: "immediate", Context0 events are sent immediately and don't need a separate flush — but you still need sdk.shutdown() for Langfuse.