All documentation

Instrument / guide 04

OpenAI, Anthropic and LangGraph

Wrap a provider client once and record model, tokens, latency, streaming and errors.

Copy or share.

OpenAI, Anthropic and LangGraph

If your agent already calls OpenAI or Anthropic, you do not need to write spans by hand. Wrap the client once and every supported call is recorded, with the same response types, streaming behavior and errors you had before. Telemetry failures never replace the provider's result or error.

TypeScript

Provider packages are optional. Install the ones you use, then wrap the client:

ts
import OpenAI from "openai";
import Anthropic from "@anthropic-ai/sdk";
import { wrapOpenAI, wrapAnthropic, TracifyLangGraphHandler, traceAgent } from "tracify-sdk";

const openai = wrapOpenAI(new OpenAI());
const anthropic = wrapAnthropic(new Anthropic());

const reply = await openai.responses.create({
  model: process.env.OPENAI_MODEL!,
  input: "Hello",
});

const stream = await anthropic.messages.create({
  model: process.env.ANTHROPIC_MODEL!,
  max_tokens: 100,
  messages: [{ role: "user", content: "Hello" }],
  stream: true,
});
for await (const event of stream) {
  // your existing handling
}

For LangGraph, pass one reusable handler in the invocation config:

ts
await graph.invoke(input, { callbacks: [new TracifyLangGraphHandler()] });

Python

Install the extras for the providers you use:

bash
pip install 'tracify-sdk[openai,anthropic,langgraph]'
python
import tracify
from openai import OpenAI, AsyncOpenAI
from anthropic import Anthropic

openai = tracify.wrap_openai(OpenAI())
anthropic = tracify.wrap_anthropic(Anthropic())
async_openai = tracify.wrap_openai(AsyncOpenAI())

reply = openai.responses.create(model=MODEL, input="Hello")

with anthropic.messages.stream(
    model=MODEL,
    max_tokens=100,
    messages=[{"role": "user", "content": "Hello"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="")

graph.invoke(inputs, config={"callbacks": [tracify.langgraph_callback()]})

What is recorded

Each call records the model, input and output, latency, token usage, time to first token for streams, and provider errors. Provider usage is stored exactly as returned, including cache breakdowns in metadata.

Supported calls:

  • OpenAI chat completions and Responses create, plus their streams.
  • Anthropic messages create and stream, plus the beta equivalents.
  • LangGraph graph, node, tool and model callbacks, with parent relationships preserved.

How runs and parents are decided

Calls made inside traceAgent or span (TypeScript), or trace_agent or span (Python), join that run under the current step. A standalone call outside any run gets its own completed root, so unrelated requests never share one run.

Cost is never guessed

Tracify does not know your negotiated prices, so cost defaults to zero and the span is marked with costKnown false. To record USD cost, give the wrapper your own current rate card:

ts
// Your own rates, in USD per token
const rates: Record<string, { input: number; output: number }> = {
  "your-model-id": { input: 0.000001, output: 0.000004 },
};

const openai = wrapOpenAI(new OpenAI(), {
  calculateCost: ({ model, inputTokens, outputTokens }) =>
    rates[model].input * inputTokens + rates[model].output * outputTokens,
});
python
# Your own rates, in USD per token: (input, output)
RATES = {"your-model-id": (0.000001, 0.000004)}

def calculate_cost(usage):
    input_rate, output_rate = RATES[usage["model"]]
    return input_rate * usage["input_tokens"] + output_rate * usage["output_tokens"]

openai = tracify.wrap_openai(OpenAI(), calculate_cost=calculate_cost)

The callback receives the provider, model, input and output tokens, and the raw usage object (as a dictionary in Python). The numbers above are placeholders, not real prices. When the provider reports no usage, token counts stay unknown rather than estimated.

Options

  • client (client= in Python): send spans through a custom TracifyClient.
  • name: override the span name.
  • captureInput and captureOutput (capture_input, capture_output): set to false to keep prompts or responses out of the trace.

Avoid double recording

Use either a framework's model callbacks or a provider wrapper for each model call, not both. Using both records the same call twice.

Stream conversion helpers such as .tee() and .toReadableStream() are passed through and are not observed. Iterate the wrapped stream itself to capture complete usage and time to first token. When you stop a stream early, close it so the partial output is recorded once.