Built In Metrics

Agent Metrics Instrumentation

You're using a skill that wires LaunchDarkly agent metrics around an existing provider call. Your job is to audit what's already there, pick the right tier from the ladder below, and implement it with the least ceremony that still captures the metrics the Monitoring tab needs (duration, input/output tokens, success/error, plus TTFT when streaming).

The single most important thing to get right: default to the highest tier that fits the shape of the call. Going lower ("just write the manual tracker calls") looks flexible but costs you drift, missed metrics, and legacy patterns the SDKs have moved past.

The four-tier ladder

This is the order the official SDK READMEs (Python core, Node core, and every provider package) recommend. Walk from the top and stop at the first tier that fits:

Every provider — OpenAI, LangChain, Vercel, Bedrock, Anthropic, Gemini, custom HTTP — uses the same generic shape: tracker.trackMetricsOf(getAIMetricsFromResponse, () => providerCall()) in Node, tracker.track_metrics_of(get_ai_metrics_from_response, provider_call) in Python. The extractor is the only thing that changes per provider: import getAIMetricsFromResponse from the matching @launchdarkly/server-sdk-ai-<provider> (or ldai_<provider>) package, or write a small custom function that returns LDAIMetrics. There are no provider-specific tracker methods.

Workflow

1. Explore the existing call site

Before picking a tier, find the provider call and answer these questions:

• [ ] Shape? Is it a chat loop (history + turn-based), a one-shot completion, an agent step, or something else? → drives Tier 1 vs 2.
• [ ] Framework? Raw provider SDK? LangChain / LangGraph? Vercel AI SDK? CrewAI? Strands? → drives which Tier-2 provider package (if any) applies.
• [ ] Provider? OpenAI, Anthropic, Bedrock, Gemini, Azure, custom HTTP? → cross-reference with the package availability matrix below.
• [ ] Streaming? If yes, you'll need TTFT tracking, which means Tier 4 for the TTFT part even if the rest is Tier 2.
• [ ] Language? Python or Node? Provider-package coverage differs between them.
• [ ] Already using a config? If not, route to configs-create first — tracking requires a tracker, which is obtained by calling create_tracker() / createTracker() on the config object returned by completion_config() / completionConfig() / createModel().
• [ ] On the current SDK API? If the call site uses aiclient.config(...) / aiClient.config(...) or constructs an AIConfig(...) / LDAIConfig default, it's on the pre-0.20 surface. Migrate it as part of this work before adding tracking:
• aiclient.config(...) → aiclient.completion_config(...) for one-shot/chat or aiclient.agent_config(...) for agent mode (mirror the call signature). Node is the same with camelCase.
• AIConfig(...) default → AICompletionConfigDefault(...) or AIAgentConfigDefault(...) (Node: LDAICompletionConfigDefault / LDAIAgentConfigDefault). AIConfig is the base class the SDK returns; it isn't a valid default-value constructor — the typed *Default variants are.
• If the result was being tuple-unpacked (config, tracker = aiclient.config(...)), drop the unpack — the new methods return a single config object. Obtain the tracker via config.create_tracker() / aiConfig.createTracker().
• For deeper rewrites (call sites with hardcoded model/prompt as well), hand off to migrate instead of doing the full migration here.

2. Look up your Tier-2 option

Use this matrix to decide whether Tier 2 (provider package) is available for your situation. If it's not, drop to Tier 3 (custom extractor). If the shape is chat-loop, go to Tier 1 first regardless of what's in this matrix.

3. Implement from the matching reference

Once you know the tier and the provider, open the reference file and follow the pattern. The references are written so Tier 1 is always the first example, Tier 2/3 next, and Tier 4 last. Stop at the first tier that matches the app's shape.

Guardrails that apply to every tier:

1. Always check `config.enabled` before making the tracked call. A disabled config means the user has flagged the feature off — you should short-circuit to whatever fallback the app uses (cached response, error, degraded path) rather than making the provider call at all. 2. Wrap the existing call, don't rewrite it. Tier 2 and Tier 3 are designed to slot around an unmodified provider call. If you find yourself rewriting the call to fit the tracker, you're at the wrong tier — drop down one. 3. Errors are handled inside `trackMetricsOf`. The wrapper catches exceptions, records trackError() internally, and re-raises — do not add except: tracker.trackError() on top, it's a noop that also trips the at-most-once guard. Tier 1 handles both paths automatically. At Tier 4 (manual, streaming, track_duration_of) the caller does own the error-tracking call. 4. Always flush before close. Call ldClient.flush() (Python: ldclient.get().flush(); Node: await ldClient.flush()) before closing the client. Trailing events are at risk of being lost otherwise — in short-lived scripts and long-running services alike. In Node, ldClient.close() returns a Promise; await it.

4. Verify

Confirm the Monitoring tab fills in:

• [ ] Run one real request through the instrumented path.
• [ ] Open the config in LaunchDarkly → Monitoring tab. Duration, token counts, and generation counts should appear within 1–2 minutes.
• [ ] Force an error (bad API key, zero max_tokens, whatever) and confirm the error count increments.
• [ ] If streaming: verify TTFT appears. If it doesn't, you probably wrapped the stream creation with trackMetricsOf but didn't add the manual trackTimeToFirstToken call — see streaming-tracking.md.

Quick reference: tracker methods

Obtain a tracker via the factory on the config object: tracker = config.create_tracker() (Python) or const tracker = aiConfig.createTracker() (Node). Call the factory once per execution and reuse the returned tracker for every call — each factory invocation mints a new runId that tags every tracking event emitted by that tracker so events from a single execution can be correlated together (via exported events / downstream systems). The Monitoring tab aggregates events rather than grouping them by run today — the runId is useful when events are exported or queried outside the UI, and is the identifier the SDK's at-most-once guards are keyed on. The methods below are the raw API surface — most of the time you should not call them individually; use trackMetricsOf or a Tier-1 managed runner. The list is here so you can recognize the methods in existing code and reach for the right one when you genuinely need Tier 4.

Related skills

• configs-create — prerequisite if the app doesn't have a config yet
• custom-metrics — business metrics (conversion, resolution, retention) layered on top of the agent metrics this skill captures
• online-evals — automatic quality scoring (LLM-as-judge) on sampled live requests; complementary to the metrics here
• migrate — Stage 4 of the hardcoded-to-AgentControl migration delegates to this skill

Anthropic Metrics Tracking

There is no LaunchDarkly provider package for Anthropic direct API today. The canonical path is the generic trackMetricsOf wrapper (Tier 3) with a small custom extractor that reads response.usage.input_tokens and response.usage.output_tokens. The instinct to "default to generic" is correct here — there is no Tier-2 shortcut to take.

Three viable paths, in order of preference:

1. Route Anthropic through LangChain. If the app already uses LangChain (or can adopt it cheaply), install the LangChain provider package and use it as Tier 2. LangChain's ChatAnthropic wrapper exposes the standardized usage_metadata that getAIMetricsFromResponse reads. 2. Route Anthropic through Bedrock Converse. If the app can switch to Bedrock Converse (Claude is available on Bedrock), you inherit Bedrock's Converse response shape and a custom-extractor pattern that's slightly cleaner. See bedrock-tracking.md. 3. Custom extractor on the direct SDK (this file's primary pattern).

Tier 1 is not available

ManagedModel does not currently ship an Anthropic provider. If you need Tier 1 for a chat app, use option 1 or 2 above — the LangChain provider package lets ManagedModel wrap a ChatAnthropic under the hood, which restores the zero-tracker-call experience.

Tier 3 — Custom extractor + trackMetricsOf (primary)

Python — direct Anthropic SDK:

import anthropic
from ldai.providers.types import LDAIMetrics, TokenUsage

client = anthropic.Anthropic()

def anthropic_extractor(response) -> LDAIMetrics:
    return LDAIMetrics(
        success=True,
        tokens=TokenUsage(
            total=response.usage.input_tokens + response.usage.output_tokens,
            input=response.usage.input_tokens,
            output=response.usage.output_tokens,
        ),
    )

def call_with_tracking(ai_config, user_prompt: str) -> str | None:
    if not ai_config.enabled:
        return None

    system_content = ai_config.messages[0].content if ai_config.messages else ""

    def call_anthropic():
        return client.messages.create(
            model=ai_config.model.name,
            max_tokens=1024,
            system=system_content,
            messages=[{"role": "user", "content": user_prompt}],
        )

    tracker = ai_config.create_tracker()
    # Exceptions are tracked automatically here: track_metrics_of catches
    # exceptions, records tracker.track_error(), and re-raises. Do NOT add
    # except: tracker.track_error() on top — it's a noop that trips the
    # at-most-once guard. Wrap in your own try/except only if you need
    # local handling (logging, fallback, alert); the error is already tracked.
    response = tracker.track_metrics_of(anthropic_extractor, call_anthropic)
    return response.content[0].text

Node — direct Anthropic SDK:

import Anthropic from '@anthropic-ai/sdk';
import type { LDAIMetrics } from '@launchdarkly/server-sdk-ai';

const client = new Anthropic();

const anthropicExtractor = (response: Anthropic.Message): LDAIMetrics => ({
  success: true,
  tokens: {
    total: response.usage.input_tokens + response.usage.output_tokens,
    input: response.usage.input_tokens,
    output: response.usage.output_tokens,
  },
});

async function callWithTracking(
  aiConfig: LDAICompletionConfig,
  userPrompt: string,
): Promise<string | null> {
  if (!aiConfig.enabled) return null;

  const systemContent = aiConfig.messages?.[0]?.content ?? '';

  const tracker = aiConfig.createTracker();
  // Exceptions are tracked automatically: trackMetricsOf catches exceptions,
  // records tracker.trackError(), and re-throws. Do NOT add
  // catch (err) { tracker.trackError(); throw err } on top — it's a noop
  // that trips the at-most-once guard. Wrap in your own try/catch only if
  // you need local handling (logging, fallback); the error is already tracked.
  const response = await tracker.trackMetricsOf(
    anthropicExtractor,
    () => client.messages.create({
      model: aiConfig.model!.name,
      max_tokens: 1024,
      system: systemContent,
      messages: [{ role: 'user', content: userPrompt }],
    }),
  );
  return response.content[0].type === 'text' ? response.content[0].text : null;
}

Notes on the extractor shape:

• Anthropic returns input_tokens / output_tokens on response.usage. Compute total yourself; Anthropic does not provide it.
• LDAIMetrics is a typed surface — Python has it at ldai.providers.types, Node exports it from @launchdarkly/server-sdk-ai. Keep the extractor pure: no side effects, no network calls.
• success: true in the extractor is not a lie — trackMetricsOf only calls the extractor on the success path. On the error path, trackMetricsOf records trackError() internally and re-throws; no caller-side catch block is required.

Tier 2 option — route via LangChain

If the app can adopt LangChain, the LangChain provider package handles Anthropic (via @langchain/anthropic) through the same trackMetricsOf(getAIMetricsFromResponse, ...) pattern used for any other LangChain model. This is often the cleanest answer if the app already uses or is open to LangChain, because the extractor is built in and shared with every other LangChain-wrapped model.

from ldai_langchain import create_langchain_model, get_ai_metrics_from_response

ai_config = ai_client.completion_config("my-config-key", context, default_config)
llm = create_langchain_model(ai_config)  # ChatAnthropic under the hood

tracker = ai_config.create_tracker()
response = tracker.track_metrics_of(
    get_ai_metrics_from_response,
    lambda: llm.invoke(messages),
)

Tier 4 — Manual (streaming only)

Streaming Anthropic needs manual TTFT tracking; the pattern is identical to OpenAI streaming. See streaming-tracking.md.

What NOT to do

• Do not hand-wire `track_duration_of` + `track_tokens` + `track_success` as three separate calls unless you're on the streaming path. That's Tier 4, and trackMetricsOf gives you the same three metrics in one call with half the drift surface.
• Do not look for a `track_anthropic_metrics` helper — it doesn't exist, never has, and won't be added. Anthropic direct support lives in the extractor you write above.
• Do not invent a provider package like @launchdarkly/server-sdk-ai-anthropic. It doesn't exist as of this writing. Check js-core ai-providers before recommending one.

AWS Bedrock Metrics Tracking

There is no LaunchDarkly provider package for Bedrock today (neither Python nor Node). Two practical paths:

1. Route Bedrock through LangChain (ChatBedrockConverse / langchain-aws). If you're open to LangChain, this is the closest thing to Tier 2 — you use the LangChain provider package's getAIMetricsFromResponse and inherit the whole trackMetricsOf pattern for free. 2. Custom extractor on `boto3` (this file's primary pattern). Bedrock Converse returns a stable response shape with usage.inputTokens / usage.outputTokens / usage.totalTokens, so the extractor is three lines.

Tier 1 is not available

ManagedModel does not ship a Bedrock provider today (Python or Node). If you want Tier 1 for a Bedrock chat app, route via LangChain — ManagedModel can wrap a ChatBedrockConverse through the LangChain provider package.

Tier 3 — Custom extractor + trackMetricsOf (primary)

Converse API (recommended)

Python:

import boto3
from ldai.providers.types import LDAIMetrics, TokenUsage

bedrock = boto3.client("bedrock-runtime")

def bedrock_converse_extractor(response) -> LDAIMetrics:
    usage = response.get("usage", {})
    return LDAIMetrics(
        success=True,
        tokens=TokenUsage(
            total=usage.get("totalTokens", 0),
            input=usage.get("inputTokens", 0),
            output=usage.get("outputTokens", 0),
        ),
    )

def call_with_tracking(ai_config, user_prompt: str) -> str | None:
    if not ai_config.enabled:
        return None

    system_content = ai_config.messages[0].content if ai_config.messages else ""

    def call_bedrock():
        kwargs = {
            "modelId": ai_config.model.name,
            "messages": [{"role": "user", "content": [{"text": user_prompt}]}],
        }
        if system_content:
            kwargs["system"] = [{"text": system_content}]
        return bedrock.converse(**kwargs)

    tracker = ai_config.create_tracker()
    # Exceptions are tracked automatically — track_metrics_of catches
    # exceptions, records tracker.track_error(), and re-raises.
    response = tracker.track_metrics_of(bedrock_converse_extractor, call_bedrock)
    return response["output"]["message"]["content"][0]["text"]

Node:

import { BedrockRuntimeClient, ConverseCommand, type ConverseCommandOutput } from '@aws-sdk/client-bedrock-runtime';
import type { LDAIMetrics } from '@launchdarkly/server-sdk-ai';

const bedrock = new BedrockRuntimeClient({});

const bedrockConverseExtractor = (response: ConverseCommandOutput): LDAIMetrics => ({
  success: true,
  tokens: {
    total: response.usage?.totalTokens ?? 0,
    input: response.usage?.inputTokens ?? 0,
    output: response.usage?.outputTokens ?? 0,
  },
});

async function callWithTracking(
  aiConfig: LDAICompletionConfig,
  userPrompt: string,
): Promise<string | null> {
  if (!aiConfig.enabled) return null;

  const systemContent = aiConfig.messages?.[0]?.content;

  const tracker = aiConfig.createTracker();
  // Exceptions are tracked automatically — trackMetricsOf catches
  // exceptions, records tracker.trackError(), and re-throws.
  const response = await tracker.trackMetricsOf(
    bedrockConverseExtractor,
    () => bedrock.send(new ConverseCommand({
      modelId: aiConfig.model!.name,
      messages: [{ role: 'user', content: [{ text: userPrompt }] }],
      ...(systemContent ? { system: [{ text: systemContent }] } : {}),
    })),
  );
  return response.output?.message?.content?.[0]?.text ?? null;
}

Legacy InvokeModel API

InvokeModel returns per-model shapes (Anthropic on Bedrock returns Anthropic's shape, Llama on Bedrock returns Meta's shape, etc.), so the extractor has to branch. Prefer Converse unless you're locked into InvokeModel by an older model that Converse doesn't support. If you must use InvokeModel, switch the extractor based on the model family:

def invoke_model_extractor(response) -> LDAIMetrics:
    body = json.loads(response["body"].read())
    # Claude on InvokeModel
    if "usage" in body:
        return LDAIMetrics(
            success=True,
            tokens=TokenUsage(
                total=body["usage"]["input_tokens"] + body["usage"]["output_tokens"],
                input=body["usage"]["input_tokens"],
                output=body["usage"]["output_tokens"],
            ),
        )
    # Llama / Titan — use the fields on the specific body shape
    # ...
    return LDAIMetrics(success=True, tokens=TokenUsage(total=0, input=0, output=0))

This is a good reason to migrate to Converse if you can.

Tier 2 option — route via LangChain

If the app uses LangChain, the LangChain provider package's ChatBedrockConverse support gives you the Tier-2 experience:

from ldai_langchain import create_langchain_model, get_ai_metrics_from_response

ai_config = ai_client.completion_config("my-config-key", context, default_config)
llm = create_langchain_model(ai_config)  # ChatBedrockConverse when provider=bedrock

tracker = ai_config.create_tracker()
response = tracker.track_metrics_of(
    get_ai_metrics_from_response,
    lambda: llm.invoke(messages),
)

LangChain normalizes the Converse response shape into AIMessage.usage_metadata, which get_ai_metrics_from_response reads — so you don't need a Bedrock-specific extractor.

Tier 4 — Manual (streaming only)

Bedrock Converse streaming (ConverseStream) needs manual TTFT tracking. The pattern is identical to OpenAI streaming. See streaming-tracking.md.

Gemini Metrics Tracking

There is no LaunchDarkly provider package for Gemini today (neither Python nor Node). The canonical path is Tier 3: a small custom extractor composed with trackMetricsOf. The Gemini response shape is stable — response.usage_metadata / response.usageMetadata carries prompt_token_count / promptTokenCount, candidates_token_count / candidatesTokenCount, and total_token_count / totalTokenCount — so the extractor is three lines.

Tier 1 is not available

ManagedModel does not currently ship a Gemini provider. If you need Tier 1 for a chat app, route via the LangChain provider package (ChatGoogleGenerativeAI under the hood), which restores the zero-tracker-call experience. See langchain-tracking.md.

Tier 3 — Custom extractor + trackMetricsOf (primary)

Gemini's API diverges from OpenAI's in three places that matter for a wrapper:

1. System messages are a top-level field. GenerateContentConfig.system_instruction / systemInstruction carries the system prompt; the contents array only holds user and model turns. You cannot put a role: "system" item in contents. 2. Assistant messages use role `model`. Convert role: "assistant" → role: "model" when mapping LD messages into Gemini's contents. 3. Parameter names differ. max_tokens on a LaunchDarkly variation (the snake_case key shown in the LD UI) becomes max_output_tokens on Python's GenerateContentConfig, or maxOutputTokens in Node. Other LD parameter names (temperature, top_p, top_k) either pass through or map with the same helper.

Two helpers absorb the divergence — a message splitter and a parameter remapper — and the metrics extractor sits on top.

Python — google-genai:

from google import genai
from google.genai.types import Content, Part, GenerateContentConfig
from ldai.providers.types import LDAIMetrics, TokenUsage

gemini_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])

def gemini_metrics(response) -> LDAIMetrics:
    usage = response.usage_metadata
    return LDAIMetrics(
        success=True,
        tokens=TokenUsage(
            total=usage.total_token_count or 0,
            input=usage.prompt_token_count or 0,
            output=usage.candidates_token_count or 0,
        ) if usage else None,
    )

def map_to_gemini_messages(ld_messages):
    """Split LD messages into (system_instruction, contents) for google-genai.
    System messages concatenate into the top-level system_instruction; user and
    assistant messages become Content items with role 'user' or 'model'."""
    system_parts: list[str] = []
    contents: list[Content] = []
    for m in ld_messages or []:
        if m.role == "system":
            system_parts.append(m.content)
        elif m.role == "user":
            contents.append(Content(role="user", parts=[Part(text=m.content)]))
        elif m.role == "assistant":
            contents.append(Content(role="model", parts=[Part(text=m.content)]))
    return (" ".join(system_parts) or None), contents

def gemini_config_kwargs(params):
    """Map config parameter names to google-genai's GenerateContentConfig.
    LaunchDarkly stores max_tokens (snake_case, matching the LD UI); Gemini's
    Python SDK expects max_output_tokens. Drop `tools` — they go on
    GenerateContentConfig.tools directly; leaving them here would double-pass."""
    mapping = {"max_tokens": "max_output_tokens"}
    return {mapping.get(k, k): v for k, v in (params or {}).items() if k != "tools"}

def call_with_tracking(ai_config, user_prompt: str) -> str | None:
    if not ai_config.enabled:
        return None

    system_instruction, contents = map_to_gemini_messages(ai_config.messages or [])
    contents.append(Content(role="user", parts=[Part(text=user_prompt)]))

    params = (ai_config.model.to_dict().get("parameters") if ai_config.model else None) or {}

    def call_gemini():
        return gemini_client.models.generate_content(
            model=ai_config.model.name,
            contents=contents,
            config=GenerateContentConfig(
                system_instruction=system_instruction,
                **gemini_config_kwargs(params),
            ),
        )

    tracker = ai_config.create_tracker()
    # Exceptions are tracked automatically — track_metrics_of catches
    # exceptions, records tracker.track_error(), and re-raises.
    response = tracker.track_metrics_of(gemini_metrics, call_gemini)
    return response.text

Node — @google/genai:

import { GoogleGenAI, type Content } from '@google/genai';
import type { LDAIMetrics } from '@launchdarkly/server-sdk-ai';

const genAI = new GoogleGenAI({ apiKey: process.env.GEMINI_API_KEY! });

const geminiMetrics = (response: any): LDAIMetrics => {
  const usage = response.usageMetadata;
  return {
    success: true,
    tokens: usage
      ? {
          total: usage.totalTokenCount ?? 0,
          input: usage.promptTokenCount ?? 0,
          output: usage.candidatesTokenCount ?? 0,
        }
      : undefined,
  };
};

function mapToGeminiMessages(
  ldMessages?: Array<{ role: string; content: string }>,
): { systemInstruction: string | undefined; contents: Content[] } {
  const contents: Content[] = [];
  const systemParts: string[] = [];
  for (const m of ldMessages ?? []) {
    if (m.role === 'system') systemParts.push(m.content);
    else if (m.role === 'user') contents.push({ role: 'user', parts: [{ text: m.content }] });
    else if (m.role === 'assistant') contents.push({ role: 'model', parts: [{ text: m.content }] });
  }
  return {
    systemInstruction: systemParts.length ? systemParts.join(' ') : undefined,
    contents,
  };
}

// Map config parameter names to @google/genai's GenerateContentConfig keys.
// LaunchDarkly stores max_tokens (snake_case, matching the LD UI); @google/genai
// expects maxOutputTokens. Drop `tools` — they go on GenerateContentConfig.tools
// directly; leaving them here would double-pass.
function geminiConfigFields(params: Record<string, unknown>): Record<string, unknown> {
  const mapping: Record<string, string> = { max_tokens: 'maxOutputTokens' };
  return Object.fromEntries(
    Object.entries(params ?? {})
      .filter(([k]) => k !== 'tools')
      .map(([k, v]) => [mapping[k] ?? k, v]),
  );
}

async function callWithTracking(
  aiConfig: LDAICompletionConfig,
  userPrompt: string,
): Promise<string | null> {
  if (!aiConfig.enabled) return null;

  const { systemInstruction, contents } = mapToGeminiMessages(aiConfig.messages);
  contents.push({ role: 'user', parts: [{ text: userPrompt }] });

  const params = (aiConfig.model?.parameters ?? {}) as Record<string, unknown>;

  const tracker = aiConfig.createTracker();
  // Exceptions are tracked automatically — trackMetricsOf catches
  // exceptions, records tracker.trackError(), and re-throws.
  const response = await tracker.trackMetricsOf(
    geminiMetrics,
    () => genAI.models.generateContent({
      model: aiConfig.model!.name,
      contents,
      config: {
        systemInstruction,
        ...geminiConfigFields(params),
      },
    }),
  );
  return response.text ?? null;
}

Notes on the extractor shape:

• Gemini uses snake_case in Python (prompt_token_count) and camelCase in Node (promptTokenCount). The LD TokenUsage / LDAIMetrics type is the same in both.
• total_token_count already includes input + output from Google; do not recompute it.
• success: true in the extractor is not a lie — trackMetricsOf only calls the extractor on the success path. On the error path, trackMetricsOf records trackError() internally and re-throws; no caller-side catch block is required.

Tools

LaunchDarkly stores attached tools on ai_config.model.parameters.tools in the flat {type, name, description, parameters} shape. Gemini's GenerateContentConfig.tools expects a list of {function_declarations: [{name, description, parameters}]} blocks (Python) or {functionDeclarations: [...]} (Node), so convert at runtime:

ld_tools = (params.get("tools") or [])
gemini_tools = [
    {
        "function_declarations": [
            {
                "name": t["name"],
                "description": t.get("description", ""),
                "parameters": t.get("parameters", {"type": "object", "properties": {}}),
            }
            for t in ld_tools
        ],
    }
] if ld_tools else []

Tool handlers stay in your application code — LaunchDarkly stores the schema, your application owns the behavior. For the full agent loop pattern (MAX_STEPS, functionCalls handling, tracker.track_tool_call), see the agent-mode section of tools.

Tier 2 option — route via LangChain

If the app can adopt LangChain, the LangChain provider package handles Gemini (via @langchain/google-genai / langchain-google-genai) through the standard trackMetricsOf(getAIMetricsFromResponse, ...) pattern. The provider package handles LaunchDarkly→LangChain provider-name mapping (for example, "gemini" → "google_genai") and forwards all variation parameters automatically, so you do not need your own mapping helper. See langchain-tracking.md.

Tier 4 — Manual (streaming only)

Streaming Gemini needs manual TTFT tracking; the pattern is identical to OpenAI streaming. See streaming-tracking.md.

What NOT to do

• Do not look for a `track_gemini_metrics` helper — it does not exist. Gemini support lives in the extractor above.
• Do not invent a provider package like @launchdarkly/server-sdk-ai-gemini or launchdarkly-server-sdk-ai-gemini. Neither exists on npm or PyPI. Check ai-providers in js-core and python-server-sdk-ai/packages/ai-providers before recommending one.
• Do not put `role: "system"` items inside `contents`. Gemini will either ignore them or error. The system prompt goes on system_instruction / systemInstruction.
• Do not assume LaunchDarkly stores `maxTokens` (camelCase) as the parameter key. The UI and the stored variation use max_tokens. The mapping helper renames it to max_output_tokens / maxOutputTokens for Gemini's SDK.

LangChain & LangGraph Metrics Tracking

LangChain is covered by a first-class LaunchDarkly provider package in both Python and Node. The same package is what LangGraph rides on — there is no separate LangGraph helper.

• Python: launchdarkly-server-sdk-ai-langchain (imported as ldai_langchain)
• Node: @launchdarkly/server-sdk-ai-langchain

Three helpers do the heavy lifting. Use them — skipping any silently drops value that the provider package would otherwise give you.

model.parameters vs model.custom — the biggest gotcha

create_langchain_model forwards every key on config.model.parameters to the underlying provider SDK via init_chat_model. That means any app-scoped knob you want to drive from LaunchDarkly — search result limits, retry budgets, feature toggles, prompt-variable defaults — must not live in parameters, because the provider will reject unknown kwargs at runtime (e.g., AsyncMessages.create() got an unexpected keyword argument 'max_search_results').

Put provider-bound fields in model.parameters and app-scoped fields in model.custom:

# Read a provider-bound parameter (forwarded to the LLM SDK)
temperature = ai_config.model.get_parameter("temperature")

# Read an app-scoped knob (NOT forwarded, safe for anything)
max_search_results = ai_config.model.get_custom("max_search_results") or 10

MCP caveat — two paths, pick one. The LaunchDarkly MCP update-ai-config-variation tool does not currently expose the top-level custom field on a variation. You have two options:

Option A — PATCH via REST API. Cleanest shape (value lands at model.custom where the Python/Node SDKs expose it via get_custom(...) / custom accessors) but requires a separate LD_API_KEY with write scope:

curl -X PATCH \
  "https://app.launchdarkly.com/api/v2/projects/$PROJECT/ai-configs/$CONFIG_KEY/variations/$VARIATION_ID" \
  -H "Authorization: $LD_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"patch":[{"op":"add","path":"/model/custom","value":{"max_search_results":10}}]}'

Option B — write via MCP under `parameters`, read via a defensive accessor. MCP does accept a custom entry inside parameters, but it lands at model.parameters.custom instead of model.custom. This shape is not what the provider SDK wants — create_langchain_model forwards every parameters key to init_chat_model, so naming the key custom at the parameters level would still get forwarded (and rejected). The workaround is to keep the shape but have the app read from both locations via a defensive accessor:

def get_custom(ai_config, key: str, default=None):
    """Read an app-scoped knob from model.custom, falling back to
    model.parameters['custom'] to cover the MCP-inserted shape.
    Remove the fallback once the MCP tool exposes top-level custom."""
    # Preferred shape (REST API / future MCP versions)
    value = ai_config.model.get_custom(key)
    if value is not None:
        return value
    # MCP fallback shape — parameters.custom as a nested dict
    params = ai_config.model.parameters or {}
    nested = params.get("custom") or {}
    return nested.get(key, default)

max_results = get_custom(ai_config, "max_search_results", default=10)

Two things to verify when using Option B: (1) the key inside parameters.custom is not passed on to the provider SDK — init_chat_model forwards parameters wholesale, so if the variation accidentally puts the knob directly in parameters (not under parameters.custom) it will still crash the provider. The nested-under-custom-dict shape is required. (2) Remove the defensive reader once MCP exposes model.custom directly — the fallback is a migration aid, not a permanent interface.

Nothing in the tracker or provider packages reads custom — it's a pass-through bucket for your application to pull from via config.model.get_custom(key) (or the defensive accessor above while the MCP gap remains).

Tier 2 — LangChain (single model, not a graph)

The common case: a one-shot LangChain call (ChatOpenAI, ChatAnthropic, ChatGoogleGenerativeAI, ChatBedrockConverse, etc.) against a config in completion mode.

Python:

from ldai_langchain import (
    create_langchain_model,
    convert_messages_to_langchain,
    get_ai_metrics_from_response,
)
from langchain_core.messages import HumanMessage

config = ai_client.completion_config("my-config-key", context)
if not config.enabled:
    return None

# create_langchain_model reads config.model.name + parameters and picks the
# right chat class (ChatOpenAI, ChatAnthropic, …) with no per-provider branching.
llm = create_langchain_model(config)

messages = convert_messages_to_langchain(config.messages or [])
messages.append(HumanMessage(content=user_prompt))

tracker = config.create_tracker()
# Exceptions are tracked automatically — track_metrics_of_async catches
# exceptions, records tracker.track_error(), and re-raises.
completion = await tracker.track_metrics_of_async(
    get_ai_metrics_from_response,
    lambda: llm.ainvoke(messages),
)
return completion.content

Node:

import {
  createLangChainModel,
  convertMessagesToLangChain,
  getAIMetricsFromResponse,
} from '@launchdarkly/server-sdk-ai-langchain';
import { HumanMessage } from '@langchain/core/messages';

const aiConfig = await aiClient.completionConfig('my-config-key', context);
if (!aiConfig.enabled) return null;

// createLangChainModel picks the right chat class (ChatOpenAI, ChatAnthropic, …)
// and forwards all variation parameters.
const llm = await createLangChainModel(aiConfig);

const messages = convertMessagesToLangChain(aiConfig.messages ?? []);
messages.push(new HumanMessage(userPrompt));

const tracker = aiConfig.createTracker();
// Exceptions are tracked automatically — trackMetricsOf catches
// exceptions, records tracker.trackError(), and re-throws.
const completion = await tracker.trackMetricsOf(
  getAIMetricsFromResponse,
  () => llm.invoke(messages),
);
return completion.content;

Both create_langchain_model (Python) and createLangChainModel (Node) raise at model-creation time if the matching LangChain provider integration is not installed. For example, if the variation's provider.name is anthropic, your environment needs langchain-anthropic (Python) or @langchain/anthropic (Node). The error surface is LangChain's, not LaunchDarkly's — install the missing integration and re-run.

Why not init_chat_model + a custom provider-name mapping helper?

You will see examples in the wild that build the model by hand with init_chat_model(model=config.model.name, model_provider=map_provider_to_langchain(config.provider.name)). Do not do this. It silently drops every parameter set on the variation (temperature, max_tokens, top_p, stop sequences, and any new field LaunchDarkly adds later), because init_chat_model only receives the name and provider. create_langchain_model forwards the whole parameter dict.

Tier 2 — LangGraph (agent workflows)

LangGraph's prebuilt agent takes a model, tools, and a system prompt. Build the model with create_langchain_model (Python) or createLangChainModel (Node) and pass it in. The tracker wraps the whole agent invocation; the extractor aggregates token usage across every message the agent produced, and tool-call telemetry is read off the result after the wrapped call returns.

API note (Python). Use from langchain.agents import create_agent. The earlier from langgraph.prebuilt import create_react_agent is deprecated in LangGraph 1.0 and removed in 2.0 — same return shape; the only call-site rename is prompt= → system_prompt=. Node still uses createReactAgent from @langchain/langgraph/prebuilt.

Python — agent mode with a MemorySaver checkpointer. The Python helper package ships sum_token_usage_from_messages (token aggregation across the agent's output messages) and get_tool_calls_from_response (tool-call name extraction per message); use them inside the track_metrics_of_async extractor / loop instead of hand-rolling either:

from ldai.providers.types import LDAIMetrics
from ldai_langchain import (
    create_langchain_model,
    get_tool_calls_from_response,
    sum_token_usage_from_messages,
)
from langchain.agents import create_agent
from langgraph.checkpoint.memory import MemorySaver

agent_config = ai_client.agent_config("my-agent-key", context)
if not agent_config.enabled:
    return None

llm = create_langchain_model(agent_config)

# MemorySaver gives the ReAct agent short-term memory per thread_id.
checkpointer = MemorySaver()
agent = create_agent(
    llm,
    [...],                                # application-owned tool handlers
    system_prompt=agent_config.instructions,
    checkpointer=checkpointer,
)

# track_metrics_of_async records duration + success/error itself; the
# extractor only returns LDAIMetrics. The surrounding try/except is for
# local logging, not for tracker bookkeeping.
tracker = agent_config.create_tracker()
try:
    result = await tracker.track_metrics_of_async(
        lambda res: LDAIMetrics(
            success=True,
            tokens=sum_token_usage_from_messages(res.get("messages", [])),
        ),
        lambda: agent.ainvoke(
            {"messages": [{"role": "user", "content": user_prompt}]},
            config={"configurable": {"thread_id": thread_id}},
        ),
    )
    for msg in result.get("messages", []):
        for name in get_tool_calls_from_response(msg):
            tracker.track_tool_call(name)
except Exception as e:
    # Already recorded by track_metrics_of_async — log locally if needed.
    raise

Node — same pattern with trackMetricsOf + a custom aggregator:

import {
  createLangChainModel,
  getAIMetricsFromResponse,
} from '@launchdarkly/server-sdk-ai-langchain';
import type { LDAIMetrics } from '@launchdarkly/server-sdk-ai';
import { createReactAgent } from '@langchain/langgraph/prebuilt';
import { MemorySaver } from '@langchain/langgraph';

const agentConfig = await aiClient.agentConfig('my-agent-key', context);
if (!agentConfig.enabled) return null;

const llm = await createLangChainModel(agentConfig);
const checkpointer = new MemorySaver();
const agent = createReactAgent({
  llm,
  tools: [/* ... */],
  prompt: agentConfig.instructions,
  checkpointer,
});

// Aggregate tokens across every message the agent produced.
const langgraphMetrics = (result: any): LDAIMetrics => {
  let input = 0, output = 0, total = 0;
  for (const message of result.messages ?? []) {
    const m = getAIMetricsFromResponse(message);
    if (m.tokens) {
      input += m.tokens.input ?? 0;
      output += m.tokens.output ?? 0;
      total += m.tokens.total ?? 0;
    }
  }
  return { success: true, tokens: total > 0 ? { input, output, total } : undefined };
};

// trackMetricsOf records duration + success/error itself; do not call
// trackError after this — it would be a redundant second event.
const agentTracker = agentConfig.createTracker();
const result = await agentTracker.trackMetricsOf(
  langgraphMetrics,
  () => agent.invoke(
    { messages: [{ role: 'user', content: userPrompt }] },
    { configurable: { thread_id: threadId } },
  ),
);

// Tool-call telemetry: walk the result messages.
for (const msg of result.messages ?? []) {
  for (const tc of (msg as any).tool_calls ?? []) {
    agentTracker.trackToolCall(tc.name);
  }
}

Why aggregate per message

get_ai_metrics_from_response / getAIMetricsFromResponse is defined on a single LangChain AIMessage. A LangGraph run produces N messages (model turn, tool result, model turn, tool result, final). If you pass the whole result to the extractor, you miss most of the token usage. Iterating and summing is deliberate — it's the same pattern the LaunchDarkly LangGraph guide uses.

Binding config-attached tools with build_structured_tools

If the variation has tools attached (via /tools), use build_structured_tools rather than hand-rolling a TOOL_REGISTRY / resolve_tools / ALL_TOOLS shape. The helper reads ai_config.model.parameters.tools, picks the matching entries from your {name: callable} registry, wraps them as LangChain StructuredTool instances, and preserves the LD tool key as the StructuredTool.name (so ToolNode(...) lookup works without a second mapping).

# tools.py — implementations only; no manual schema, no resolve_tools()
from langchain_tavily import TavilySearch

async def search(query: str) -> dict:
    """Search the web via Tavily."""
    ai_config = get_agent_config(...)
    max_results = ai_config.model.get_custom("max_search_results") or 10
    return await TavilySearch(max_results=max_results).ainvoke({"query": query})

TOOL_REGISTRY = {"search": search}

# graph.py — bind whatever the active variation exposes
from ldai_langchain import create_langchain_model, get_ai_metrics_from_response
from ldai_langchain.langchain_helper import build_structured_tools

model = create_langchain_model(ai_config)
tools = build_structured_tools(ai_config, TOOL_REGISTRY)
response = await tracker.track_metrics_of_async(
    get_ai_metrics_from_response,
    lambda: model.bind_tools(tools).ainvoke(messages),
)

What you delete when you adopt this: any module-level ALL_TOOLS list, any resolve_tools(tool_keys) helper, any hand-written JSON Schema blocks in code. The variation owns the schema; your repo owns the behavior. ToolNode can be seeded with every callable in the registry because the LLM only sees the filtered subset build_structured_tools produces.

Tier 3 — fall through to a custom extractor

You will not usually need Tier 3 for LangChain or LangGraph — get_ai_metrics_from_response normalizes the response shape across providers. If the variation points at a model whose LangChain integration does not populate usage_metadata (rare, usually a custom integration), write a small extractor that reads whatever field the integration exposes and returns LDAIMetrics. This is the same fallback documented in openai-tracking.md and anthropic-tracking.md.

Tier 4 — Manual (streaming only)

LangChain streaming with TTFT tracking uses the same manual pattern as direct-SDK streaming. See streaming-tracking.md.

What NOT to do

• Do not build the model with `init_chat_model` + a hand-rolled provider-name mapping. The helper forwards all variation parameters; the hand-rolled version silently drops them.
• If an existing `load_chat_model` / `init_chat_model` wrapper is already in the repo — delete it. Do not keep it around as a convenience. Leaving it in place means every future agent in the codebase will reach for the familiar function and silently drop variation parameters. Replace imports with create_langchain_model(ai_config) at every call site, then remove the wrapper file.
• Do not keep hand-rolled `TOOL_REGISTRY` / `resolve_tools` / `ALL_TOOLS` patterns once `build_structured_tools` is available. The SDK helper replaces them. Same deletion principle as above — if a hand-rolled version sits in the repo, future code will use it instead of the SDK helper.
• Do not put app-scoped knobs in `model.parameters`. They will be forwarded to the provider SDK and crash at runtime with an unexpected-keyword-argument error. Use model.custom for anything that is not a provider-bound parameter.
• Do not pass the full LangGraph `result` object to `get_ai_metrics_from_response`. The extractor is defined on a single message; aggregating across result.messages is the correct pattern.
• Do not assume there is a separate LangGraph provider package. There is not. @launchdarkly/server-sdk-ai-langchain and ldai_langchain cover both.
• Do not import `LaunchDarklyCallbackHandler` from `ldai.langchain`. Neither the class nor the dotted module path exists in the Python package. Use the helpers above.
• Do not re-encode tool schemas inside the fallback. If LaunchDarkly is unreachable, the fallback should run without tools (or with the minimum provider-bound parameters the app needs). Putting a full tools array back into the fallback re-introduces the hardcoded config the migration was supposed to eliminate.

Metrics API

Retrieve config metrics via the LaunchDarkly REST API.

Endpoint

GET /api/v2/projects/{projectKey}/ai-configs/{configKey}/metrics

Authentication

Requires an API token with ai-configs:read permission.

headers = {
    "Authorization": "your-api-token",
    "LD-API-Version": "beta"
}

Implementation

import requests
import time
import os

def get_ai_config_metrics(project_key: str, config_key: str, env: str = "production", hours: int = 24):
    """Get config metrics for the last N hours."""
    API_TOKEN = os.environ.get("LAUNCHDARKLY_API_TOKEN")

    now = int(time.time())
    start = now - (hours * 3600)

    url = f"https://app.launchdarkly.com/api/v2/projects/{project_key}/ai-configs/{config_key}/metrics"

    params = {
        "from": start,
        "to": now,
        "env": env
    }

    headers = {
        "Authorization": API_TOKEN,
        "LD-API-Version": "beta"
    }

    response = requests.get(url, headers=headers, params=params)

    if response.status_code == 200:
        metrics = response.json()
        print(f"[OK] Metrics for {config_key} (last {hours} hours, {env}):")
        print(f"     Generations: {metrics.get('generationCount', 0):,}")
        print(f"     Success: {metrics.get('generationSuccessCount', 0):,}")
        print(f"     Errors: {metrics.get('generationErrorCount', 0):,}")
        print(f"     Input Tokens: {metrics.get('inputTokens', 0):,}")
        print(f"     Output Tokens: {metrics.get('outputTokens', 0):,}")
        print(f"     Total Tokens: {metrics.get('totalTokens', 0):,}")
        print(f"     Input Cost: ${metrics.get('inputCost', 0):.4f}")
        print(f"     Output Cost: ${metrics.get('outputCost', 0):.4f}")
        print(f"     Duration (ms): {metrics.get('durationMs', 0):,}")
        print(f"     TTFT (ms): {metrics.get('timeToFirstTokenMs', 0):,}")
        print(f"     Thumbs Up: {metrics.get('thumbsUp', 0)}")
        print(f"     Thumbs Down: {metrics.get('thumbsDown', 0)}")
        return metrics
    else:
        print(f"[ERROR] Failed to get metrics: {response.status_code}")
        return None

Response Fields

Query Parameters

Notes

• Time range is specified in Unix timestamps (seconds)
• Costs are calculated based on model pricing and token usage
• Feedback counts require user feedback tracking implementation
• Rate limits apply; see API documentation for details

OpenAI Metrics Tracking

OpenAI is covered by a first-class LaunchDarkly provider package in both Python and Node. Walk the tiers from top to bottom and stop at the first one that fits the call shape.

Tier 1 — Managed runner (chat apps)

The simplest path for conversational OpenAI calls. Zero tracker calls — duration, tokens, and success/error are all captured by run().

Python — ManagedModel via ai_client.create_model():

from ldclient import Context
from ldai import LDAIClient, AICompletionConfigDefault, ModelConfig, LDMessage, ProviderConfig

default_config = AICompletionConfigDefault(
    enabled=True,
    model=ModelConfig(name="gpt-4o"),
    provider=ProviderConfig(name="openai"),
    messages=[LDMessage(role="system", content="You are a helpful assistant.")],
)

async def handle_turn(ai_client: LDAIClient, context: Context, user_input: str) -> str:
    model = await ai_client.create_model(
        "customer-support-chat",
        context,
        default_config,
    )
    if not model:
        return "Feature is currently unavailable."
    response = await model.run(user_input)
    return response.content

Node — ManagedModel via aiClient.createModel():

import { init } from '@launchdarkly/node-server-sdk';
import { initAi } from '@launchdarkly/server-sdk-ai';

const ldClient = init(process.env.LD_SDK_KEY!);
const aiClient = initAi(ldClient);

async function handleTurn(context: LDContext, userInput: string): Promise<string> {
  const model = await aiClient.createModel(
    'customer-support-chat',
    context,
    {
      enabled: true,
      model: { name: 'gpt-4o' },
      provider: { name: 'openai' },
      messages: [{ role: 'system', content: 'You are a helpful assistant.' }],
    },
  );
  if (!model) return 'Feature is currently unavailable.';
  const response = await model.run(userInput);
  return response.content;
}

Tracking is handled inside run(). You do not need trackMetricsOf, trackSuccess, or trackTokens at this tier.

Tier 2 — Provider package + trackMetricsOf (non-chat shapes)

Use this when the call isn't a chat loop (one-shot completion, structured output, batch job, agent step). The provider package exposes a static getAIMetricsFromResponse that knows how to pull tokens out of an OpenAI response; you compose it with the generic trackMetricsOf wrapper.

Python — launchdarkly-server-sdk-ai-openai:

managed = await ai_client.create_model("my-config-key", context, default_config)
if managed:
    result = await managed.run(user_prompt)
    return result.content

managed.run() tracks automatically — the managed runner handles duration, tokens, and success/error end-to-end. If you need finer-grained control (e.g., you want to supply your own OpenAI client with custom retries), use the raw SDK + track_metrics_of with the bare extractor:

import openai
from ldai_openai import get_ai_metrics_from_response

client = openai.OpenAI()

ai_config = ai_client.completion_config("my-config-key", context, default_config)
if not ai_config.enabled:
    return None

tracker = ai_config.create_tracker()

def call_openai():
    return client.chat.completions.create(
        model=ai_config.model.name,
        messages=[
            {"role": "system", "content": ai_config.messages[0].content},
            {"role": "user", "content": user_prompt},
        ],
    )

response = tracker.track_metrics_of(get_ai_metrics_from_response, call_openai)
return response.choices[0].message.content

Node — @launchdarkly/server-sdk-ai-openai:

import { OpenAI } from 'openai';
import { getAIMetricsFromResponse } from '@launchdarkly/server-sdk-ai-openai';

const client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

const aiConfig = await aiClient.completionConfig('my-config-key', context, defaultConfig);
if (!aiConfig.enabled) return null;

const tracker = aiConfig.createTracker();
const response = await tracker.trackMetricsOf(
  getAIMetricsFromResponse,
  () => client.chat.completions.create({
    model: aiConfig.model!.name,
    messages: [
      ...aiConfig.messages,
      { role: 'user', content: userPrompt },
    ],
  }),
);
return response.choices[0].message.content;

Error handling. trackMetricsOf catches exceptions internally, records trackError() on the tracker, and re-throws — so you do not need a try/catch block that calls trackError() yourself. Call the wrapper directly; if the caller wants to log or handle the exception, do that in addition to (not instead of) letting it propagate:

const tracker = aiConfig.createTracker();
const response = await tracker.trackMetricsOf(
  getAIMetricsFromResponse,
  () => client.chat.completions.create({ /* ... */ }),
);
return response.choices[0].message.content;

Python behaves the same with track_metrics_of. Do not add except: tracker.track_error() on top — it's a noop that would also trip the at-most-once guard.

Tier 3 — Custom extractor (fallback)

You should not need Tier 3 for OpenAI — the provider package covers it. If you're using a fork, a drop-in replacement (LiteLLM, Azure OpenAI via raw HTTP), or something the provider package doesn't recognize, write a small extractor:

from ldai.providers.types import LDAIMetrics, TokenUsage

def my_openai_extractor(response) -> LDAIMetrics:
    return LDAIMetrics(
        success=True,
        tokens=TokenUsage(
            total=response.usage.total_tokens,
            input=response.usage.prompt_tokens,
            output=response.usage.completion_tokens,
        ),
    )

tracker = ai_config.create_tracker()
response = tracker.track_metrics_of(my_openai_extractor, call_openai)

Tier 4 — Manual (streaming only)

For OpenAI streaming calls you need manual tracking because the current provider packages don't capture TTFT. See streaming-tracking.md for the full pattern. The short version: the helper that looks like it should work (trackStreamMetricsOf in Node) captures tokens from stream chunks but does not record TTFT, so you still need a manual trackTimeToFirstToken call on the first content chunk.

Strands Agents Metrics Tracking

There is no LaunchDarkly provider package for Strands. Strands is a provider-agnostic agent SDK — the same Agent class runs against Anthropic, OpenAI, and Bedrock by swapping the model argument — so the tracking pattern plugs in at the agent layer, not the provider layer. Tier 3 (custom extractor + trackMetricsOf) is the canonical path.

The Strands AgentResult object exposes a metrics.accumulated_usage dict (Python) / metrics.accumulatedUsage object (Node) that already aggregates token counts across every provider call the agent made in a single invoke_async turn — including any tool-calling round trips. That means one extractor call covers the whole turn, unlike the per-response shape from Anthropic or OpenAI direct.

The key names inside accumulated_usage are camelCase even in Python: inputTokens, outputTokens, totalTokens.

Tier 1 is not available

ManagedModel does not currently ship a Strands runner. Strands owns its own agent loop and short-term memory (SlidingWindowConversationManager), so wrapping it in a LaunchDarkly managed runner would fight against the framework. Stay on Tier 3.

Tier 3 — Explicit track_duration_of + manual track_tokens (primary)

This is the shape in the LaunchDarkly Strands integration guide. Use it when the call site is already async and you want token extraction split out from duration tracking.

from ldai.tracker import TokenUsage


def track_strands_metrics(tracker, result):
    """Record token usage from a Strands AgentResult on the LD tracker."""
    usage = getattr(result.metrics, "accumulated_usage", {}) or {}
    input_tokens = usage.get("inputTokens", 0)
    output_tokens = usage.get("outputTokens", 0)
    total = usage.get("totalTokens", 0) or (input_tokens + output_tokens)
    if total > 0:
        tracker.track_tokens(
            TokenUsage(input=input_tokens, output=output_tokens, total=total)
        )


async def run_turn(agent, tracker, user_input):
    try:
        result = await tracker.track_duration_of(lambda: agent.invoke_async(user_input))
        tracker.track_success()
        track_strands_metrics(tracker, result)
        return result.message["content"][0]["text"]
    except Exception:
        tracker.track_error()
        raise

What this tracks:

• Duration — from the track_duration_of wrapper around invoke_async.
• Tokens — from accumulated_usage, including any tool-calling round trips inside the turn.
• Success / error — explicit, in the try/except.

Tier 3 — Single-call track_metrics_of_async variant

If you prefer the single-call form that matches the rest of the provider-tracking references, fold the extractor into an LDAIMetrics return and use track_metrics_of_async:

from ldai.providers.types import LDAIMetrics, TokenUsage


def strands_extractor(result) -> LDAIMetrics:
    usage = getattr(result.metrics, "accumulated_usage", {}) or {}
    input_tokens = usage.get("inputTokens", 0)
    output_tokens = usage.get("outputTokens", 0)
    total = usage.get("totalTokens", 0) or (input_tokens + output_tokens)
    return LDAIMetrics(
        success=True,
        tokens=TokenUsage(input=input_tokens, output=output_tokens, total=total),
    )


async def run_turn(agent, tracker, user_input):
    # Exceptions are tracked automatically — track_metrics_of_async catches
    # exceptions, records tracker.track_error(), and re-raises.
    result = await tracker.track_metrics_of_async(
        strands_extractor,
        lambda: agent.invoke_async(user_input),
    )
    return result.message["content"][0]["text"]

Pick the style that matches the rest of the codebase — the two variants record the same metrics.

Provider dispatch stays in your code

Strands model classes are provider-specific (AnthropicModel, OpenAIModel, BedrockModel). To serve more than one provider from a single config key, dispatch on agent_config.provider.name before constructing the Agent. See agent-mode-frameworks.md § Strands Agent for the create_strands_model dispatcher, including the rule that parameters.tools must be dropped before being passed into the Strands model class (tools flow through the Agent constructor, not through model params).

Always flush before exit

Strands examples are commonly short-lived scripts (python run_agent.py ...). Trailing analytics events can be lost if the client closes before flushing. Always call ldclient.get().flush() (and ldclient.get().close() on exit) after the last turn.

Node / TypeScript caveat

The Strands TypeScript SDK ships BedrockModel and OpenAIModel only — no AnthropicModel. The same Tier-3 pattern applies (custom extractor over result.metrics.accumulatedUsage, then tracker.trackMetricsOf or explicit trackDurationOf + trackTokens), but multi-provider variations that include Anthropic require the Python SDK today.

Streaming Metrics Tracking

This is Tier 4 — the manual fallback. Streaming is the one case where no current helper captures everything you need. The Node SDK ships trackStreamMetricsOf, which can pull tokens from stream chunks, but it does not capture time-to-first-token (TTFT). Python doesn't have a streaming helper at all. So if you want TTFT in the Monitoring tab, you have to wire it manually — and since TTFT is the whole point of streaming observability, this is almost always what you want.

If the app doesn't need TTFT (you just want total duration + tokens + success), you can use Tier 2 / Tier 3 patterns in Node via trackStreamMetricsOf, and Tier 3 in Python by consuming the whole stream into a response object and then calling trackMetricsOf on the assembled result. TTFT is the tiebreaker that forces Tier 4.

What you track

• Time to first token (TTFT) — measured from "stream request sent" to "first content chunk received."
• Total duration — measured from "stream request sent" to "stream fully consumed."
• Tokens — read from the final stream event (if the provider includes usage) or from tiktoken / provider-native counters if not.
• Success / error — explicit calls in the consumer loop.

Python — OpenAI streaming

import time
import openai
from ldai.tracker import TokenUsage

def call_streaming_with_tracking(ai_config, user_prompt: str) -> str | None:
    if not ai_config.enabled:
        return None

    tracker = ai_config.create_tracker()
    start_time = time.time()
    first_token_time = None

    try:
        stream = openai.chat.completions.create(
            model=ai_config.model.name,
            messages=[
                {"role": "system", "content": ai_config.messages[0].content},
                {"role": "user", "content": user_prompt},
            ],
            stream=True,
            stream_options={"include_usage": True},  # Required to get usage in final chunk
        )

        response_text = ""
        final_usage = None
        for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                if first_token_time is None:
                    first_token_time = time.time()
                    tracker.track_time_to_first_token(
                        int((first_token_time - start_time) * 1000)
                    )
                response_text += chunk.choices[0].delta.content
            if getattr(chunk, "usage", None):
                final_usage = chunk.usage

        tracker.track_duration(int((time.time() - start_time) * 1000))
        tracker.track_success()

        if final_usage:
            tracker.track_tokens(TokenUsage(
                total=final_usage.total_tokens,
                input=final_usage.prompt_tokens,
                output=final_usage.completion_tokens,
            ))

        return response_text

    except Exception:
        tracker.track_error()
        raise

The stream_options={"include_usage": True} flag is required — without it, OpenAI streaming does not include usage data and you fall back to tiktoken estimation.

Python — tiktoken fallback

If you can't set include_usage (older SDK, Azure OpenAI on an endpoint that doesn't support it), count tokens locally with tiktoken:

import tiktoken
from ldai.tracker import TokenUsage

def estimate_tokens(model_name: str, prompt: str, response: str) -> TokenUsage:
    try:
        enc = tiktoken.encoding_for_model(model_name)
    except KeyError:
        enc = tiktoken.get_encoding("cl100k_base")
    input_tokens = len(enc.encode(prompt))
    output_tokens = len(enc.encode(response))
    return TokenUsage(
        total=input_tokens + output_tokens,
        input=input_tokens,
        output=output_tokens,
    )

Drop it into the streaming consumer where final_usage would have been.

Node — OpenAI streaming with manual TTFT

import { OpenAI } from 'openai';

const client = new OpenAI();

async function callStreamingWithTracking(
  aiConfig: LDAICompletionConfig,
  userPrompt: string,
): Promise<string | null> {
  if (!aiConfig.enabled) return null;

  const tracker = aiConfig.createTracker();
  const startTime = Date.now();
  let firstTokenTime: number | null = null;

  try {
    const stream = await client.chat.completions.create({
      model: aiConfig.model!.name,
      messages: [
        ...aiConfig.messages,
        { role: 'user', content: userPrompt },
      ],
      stream: true,
      stream_options: { include_usage: true },
    });

    let responseText = '';
    let finalUsage: OpenAI.CompletionUsage | undefined;

    for await (const chunk of stream) {
      const delta = chunk.choices[0]?.delta?.content;
      if (delta) {
        if (firstTokenTime === null) {
          firstTokenTime = Date.now();
          tracker.trackTimeToFirstToken(firstTokenTime - startTime);
        }
        responseText += delta;
      }
      if (chunk.usage) {
        finalUsage = chunk.usage;
      }
    }

    tracker.trackDuration(Date.now() - startTime);
    tracker.trackSuccess();

    if (finalUsage) {
      tracker.trackTokens({
        total: finalUsage.total_tokens,
        input: finalUsage.prompt_tokens,
        output: finalUsage.completion_tokens,
      });
    }

    return responseText;
  } catch (err) {
    tracker.trackError();
    throw err;
  }
}

Node — trackStreamMetricsOf (no TTFT)

If the app doesn't need TTFT, the Node SDK has a built-in streaming wrapper that handles tokens + success/error + duration:

const tracker = aiConfig.createTracker();
const response = await tracker.trackStreamMetricsOf(
  (chunks) => {
    // Extract usage from the final chunk
    const final = chunks[chunks.length - 1];
    return {
      success: true,
      tokens: {
        total: final.usage?.total_tokens ?? 0,
        input: final.usage?.prompt_tokens ?? 0,
        output: final.usage?.completion_tokens ?? 0,
      },
    };
  },
  () => client.chat.completions.create({ /* ... */, stream: true, stream_options: { include_usage: true } }),
);

This is cleaner when TTFT doesn't matter (batch processing, log summarization, tasks where latency-to-first-byte isn't user-facing). If the user is going to look at the Monitoring tab's TTFT chart, though, you need the manual pattern above.

What to avoid

• Do not wrap `openai.chat.completions.create(stream=True)` with `trackMetricsOf`. It'll record duration as the time to get the stream object, not the time to consume it — and tokens won't be captured at all because the extractor sees a stream object, not a response with usage.
• Do not forget `track_success()` / `trackSuccess()`. Unlike trackMetricsOf, the manual pattern doesn't call it for you. If you skip it, the Monitoring tab won't count the generation.
• *Do not set `first_token_time` on the first chunk.* Set it on the first chunk with non-empty delta.content. Many providers emit a role/metadata chunk before the first content chunk.