/migrate

Agent-Mode Frameworks

How to wire a config in agent mode into the frameworks that take a goal/instructions string: LangGraph, CrewAI, Strands, and custom ReAct loops. Also covers the dynamic tool loading pattern from the devrel-agents-tutorial — how to extract tool names from config.tools at runtime and instantiate the actual tool implementations without hardcoding.

When to pick agent mode

Completion mode is the default and covers direct provider calls (OpenAI, Anthropic, Bedrock) where the app assembles a messages array. Pick agent mode when:

Agent mode returns an instructions string. Completion mode returns a messages array. Both modes support tools, parameters, and the same tracker — the only difference is the input shape the SDK returns to you.

Caveat: judges cannot be attached to agent-mode variations via the LaunchDarkly UI. Agent mode evaluations must go through the programmatic judge API (create_judge(...).evaluate(input, output)). See online-evals for the programmatic path.

Model construction for LangChain / LangGraph. When the framework runs on top of LangChain (which includes LangGraph's prebuilt agent and most custom graphs), build the chat model with create_langchain_model(ai_config) (Python) or createLangChainModel(aiConfig) (Node). These helpers forward every variation parameter (temperature, max_tokens, top_p, …) and handle LaunchDarkly→LangChain provider-name mapping internally. Do not hand-roll init_chat_model(model=..., model_provider=...) — it silently drops every variation parameter. See langchain-tracking.md for the canonical single-model and LangGraph patterns, including the SDK helpers sum_token_usage_from_messages / get_tool_calls_from_response (Python, ldai_langchain) used inside the track_metrics_of_async / trackMetricsOf extractor.

Framework-agnostic invariants for the run-scoped pattern

The concrete examples below use specific frameworks (LangGraph, CrewAI, Strands) and specific node names (setup_run, call_model, finalize). Treat those as incidentals. The three invariants below apply to any agent framework — DSPy, AutoGen, Pydantic AI, Haystack, LlamaIndex agents, or a hand-rolled tool loop in pure Python/TypeScript. If the framework has its own idioms, translate these three rules onto them:

1. Resolve agent_config() / agentConfig() once per user turn. Every call is a flag evaluation and emits a $ld:ai:agent:config event. Re-fetching inside a loop step or a tool body amplifies the event count per turn and lets a mid-turn targeting change swap the variation between LLM calls. Do the resolve at the highest scope that corresponds to "one user-input-to-final-response cycle" — a handler function, a LangGraph entry node, a CrewAI kickoff, whatever the framework exposes.
2. Mint one tracker via create_tracker() / createTracker() per user turn. Same scope as the agent_config call. The runId ties every event from one turn together; per-step factory calls fragment the correlation and reset the SDK's at-most-once guards. If the framework has a multi-turn session (chat thread), each turn inside the session still gets its own fresh tracker — sessions share a thread_id, not a runId.
3. Emit the five at-most-once methods once at the end of the turn. track_duration / track_tokens / track_success / track_error / track_time_to_first_token each fire at most once per tracker. Accumulate inside the loop body (sum token usage across steps, stash a perf_counter_ns timer up top), emit once after the loop exits or in a dedicated finalize node. track_tool_calls / track_feedback / track_judge_result are per-event — call them as many times as the agent does those things.

What "one user turn" means differs by app shape:

If you can answer "what's the smallest unit at which I'd want to see a single 'execution' in the Monitoring tab?" — that's the turn. Mint exactly one tracker there.

Tool-scoped and app-scoped knobs go in model.custom

If the Stage 1 audit identified configuration that isn't a native model parameter — the kind of thing a provider SDK will reject with unexpected keyword argument if you forward it — these fields belong in ModelConfig(custom={...}), not ModelConfig(parameters={...}). Typical examples:

• max_search_results (tool behavior — how many hits the search tool returns)
• chunk_size / chunk_overlap (RAG preprocessing knobs)
• retry_budget / retry_backoff (app-level retry policy)
• enable_reranking, use_cache, any boolean feature toggle the agent consumes
• any value that governs tool behavior or app behavior rather than model behavior

create_langchain_model (Python) / createLangChainModel (Node) forwards every key in parameters wholesale to the provider SDK. Anthropic, OpenAI, and Gemini all raise on unknown kwargs — a max_search_results entry in parameters crashes the request with AsyncMessages.create() got an unexpected keyword argument 'max_search_results'. Put the same field in custom and the helper leaves it alone; the app reads it where it's needed.

# Fallback: mirror the hardcoded knob shape using custom
FALLBACK = AIAgentConfigDefault(
    enabled=True,
    model=ModelConfig(
        name="claude-sonnet-4-5-20250929",
        parameters={"temperature": 0.3, "max_tokens": 2000},  # provider-bound
        custom={"max_search_results": 10},                    # app-scoped
    ),
    provider=ProviderConfig(name="anthropic"),
    instructions="You are a helpful assistant.",
)

# In the tool or app code that needs the knob:
def search(query: str) -> dict:
    ai_config = get_current_agent_config()
    max_results = ai_config.model.get_custom("max_search_results") or 10
    return TavilySearch(max_results=max_results).invoke({"query": query})

Mirror the same shape on the LaunchDarkly variation. MCP caveat. The update-ai-config-variation MCP tool does not currently expose the custom field — to populate model.custom on an existing variation, PATCH it through the REST API directly:

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

Getting knobs into tools: tool factories that close over per-run config (default)

Tools need read access to model.custom at call time. The correct pattern is a tool factory that takes the per-run ai_config resolved in setup_run (or at the top of a custom ReAct loop) and returns a tool callable that closes over whatever knobs it needs:

# tools.py
def make_search(ai_config) -> Callable[..., Any]:
    max_results = ai_config.model.get_custom("max_search_results") or 10

    async def search(query: str) -> dict:
        """Search the web for current information on a given topic."""
        return await TavilySearch(max_results=max_results).ainvoke({"query": query})

    return search

TOOL_FACTORIES = {"search": make_search}

# graph.py setup_run
built = {name: fn(ai_config) for name, fn in TOOL_FACTORIES.items()}
tools = build_structured_tools(ai_config, built)

This is the default because it has three properties nothing else preserves all at once:

1. Turn-level atomicity. A mid-turn flag change doesn't swap max_search_results between the first tool call and the second — the factory captured it once, at setup_run.
2. No extra LD evaluations. agent_config() is a flag evaluation and an emitted event. Calling it from a tool once per user turn is fine; calling it per tool invocation on a chatty agent inflates $ld:ai:agent:config counts by a factor of however many tool calls run per turn.
3. Tools don't take a dependency on LD. The tool function is a plain callable. Testing is substitution, not monkeypatching get_agent_config.

Do not call get_agent_config() (or ai_client.agent_config(...)) from inside a tool body. The alternatives — re-resolving from inside the tool, or reaching into runtime.context from inside the tool — either break turn atomicity or require plumbing LangGraph's context through every tool signature. Tool factories sidestep both problems and are also how ldai_langchain.langchain_helper.build_structured_tools expects its registry to be shaped (a dict of {name: Callable} ready to bind).

The only legitimate reason to re-fetch inside a tool is if the tool's behavior needs to follow a flag change within a single turn. That's vanishingly rare; default to factories and treat re-fetch as an exception that requires a concrete reason.

Wiring agent_config into each framework

LangGraph prebuilt agent (Python — langchain.agents.create_agent)

API note. 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.js still uses createReactAgent from @langchain/langgraph/prebuilt.

from langchain.agents import create_agent
from ldai_langchain import create_langchain_model
from ldai.client import LDAIClient, AIAgentConfigDefault, ModelConfig, ProviderConfig

FALLBACK = AIAgentConfigDefault(
    enabled=True,
    model=ModelConfig(name="gpt-4o", parameters={"temperature": 0.3}),
    provider=ProviderConfig(name="openai"),
    instructions="You are a helpful assistant.",
)

def build_agent(ai_client: LDAIClient, user_id: str, tools: list):
    context = Context.builder(user_id).kind("user").build()
    config = ai_client.agent_config("support-agent", context, FALLBACK)

    if not config.enabled:
        return None, None

    # create_langchain_model forwards every variation parameter — do NOT hand-roll
    # ChatOpenAI(model=..., temperature=...). It silently drops unnamed parameters.
    llm = create_langchain_model(config)

    agent = create_agent(
        llm,
        tools,
        system_prompt=config.instructions,
    )
    return agent, config.create_tracker()

Key points:

• system_prompt=config.instructions — the instructions string replaces the hardcoded prompt
• Model + parameters come from create_langchain_model(config) — forwards the whole model.parameters dict
• A fresh tracker is minted via config.create_tracker() and returned alongside the agent so the caller can wire Stage 4 tracking around agent.invoke(...). Each call to create_tracker() produces a new runId; the caller should treat the returned tracker as owning the execution.

CrewAI Agent

from crewai import Agent
from ldai.client import LDAIClient, AIAgentConfigDefault

def build_crew_agent(ai_client, user_id: str):
    context = Context.builder(user_id).kind("user").build()
    config = ai_client.agent_config("researcher-agent", context, FALLBACK)

    if not config.enabled:
        return None

    # CrewAI expects role/goal/backstory — split the instructions or store them
    # in the Config as three variables and pipe them in at runtime.
    return Agent(
        role="Research Analyst",
        goal="Produce a summary of recent config adoption patterns.",
        backstory=config.instructions,
        llm=config.model.name,  # CrewAI accepts a string or a LangChain model
    )

Pattern note: CrewAI's Agent takes three separate fields. If you want to drive all three from LaunchDarkly, either:

• Use prompt variables on the config ({{role}}, {{goal}}, {{backstory}}) and pass them as the variables argument to agent_config(...)
• Or store a structured JSON blob in instructions and parse it in the app

Prompt variables are cleaner and keep the config human-readable in the UI.

Strands Agent

Strands is a provider-agnostic, async-first agent SDK. The same Agent class runs against Anthropic, OpenAI, and Bedrock by swapping the model argument; tools are plain @tool-decorated Python functions passed through the constructor; and SlidingWindowConversationManager keeps short-term memory across invoke_async turns without external state. Agent-mode instructions maps directly to Agent(system_prompt=...).

Strands does not ship a first-party LaunchDarkly provider package. To serve multiple providers from a single config key, dispatch on agent_config.provider.name and construct the matching Strands model class.

Provider dispatcher. Drop parameters.tools before passing params into the Strands model class — LaunchDarkly surfaces attached tools via a flat parameters.tools shape in the variation payload, but Strands receives tools via the Agent constructor. Passing tools through a second time via model params is an error.

from strands.models.anthropic import AnthropicModel
from strands.models.openai import OpenAIModel


def create_strands_model(agent_config):
    """Map an LDAIAgentConfig to the matching Strands model class by provider."""
    provider = (agent_config.provider.name if agent_config.provider else "").lower()
    model_id = agent_config.model.name
    params = dict(agent_config.model.to_dict().get("parameters") or {})
    # LD surfaces attached tools via parameters.tools; Strands takes tools via
    # Agent(tools=[...]). Drop the key before passing params to the model class.
    params.pop("tools", None)

    if provider == "anthropic":
        # AnthropicModel requires max_tokens as a kwarg, not inside params.
        max_tokens = int(
            params.pop("max_tokens", None) or params.pop("maxTokens", None) or 1024
        )
        return AnthropicModel(model_id=model_id, max_tokens=max_tokens, params=params or None)
    if provider == "openai":
        # Pass parameters through as-is — gpt-5 wants max_completion_tokens,
        # gpt-4o wants max_tokens. Keep that choice in the LD variation.
        return OpenAIModel(model_id=model_id, params=params)
    raise ValueError(f"Unsupported provider for Strands: {provider!r}")

Call site. Build the agent once per request, pull the tracker off the config, and wrap invoke_async with track_duration_of — Strands is Tier 3 (custom extractor) because there is no provider package.

from strands import Agent, tool
from strands.agent.conversation_manager.sliding_window_conversation_manager import (
    SlidingWindowConversationManager,
)
from ldai.client import AIAgentConfigDefault, ModelConfig, ProviderConfig
from ldai.tracker import TokenUsage
from ldclient import Context


@tool
def get_order_status(order_id: str) -> str:
    """Look up the status of a customer order by order ID."""
    ...


FALLBACK = AIAgentConfigDefault(
    enabled=True,
    model=ModelConfig(name="gpt-5", parameters={"max_completion_tokens": 2000}),
    provider=ProviderConfig(name="openai"),
    instructions="You are a helpful assistant.",
)


def track_strands_metrics(tracker, result):
    """Record token usage from a Strands AgentResult on the LD tracker.

    accumulated_usage aggregates tokens across every provider call in the turn,
    including tool-calling round trips — unlike the single-response shape from
    Anthropic or OpenAI direct.
    """
    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(ai_client, user_id: str, user_input: str):
    context = Context.builder(user_id).kind("user").build()
    agent_config = ai_client.agent_config("strands-agent", context, FALLBACK)

    if not agent_config.enabled:
        return disabled_response()

    agent = Agent(
        name="order-assistant",
        model=create_strands_model(agent_config),
        system_prompt=agent_config.instructions,
        tools=[get_order_status],
        conversation_manager=SlidingWindowConversationManager(window_size=40),
    )
    tracker = agent_config.create_tracker()

    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

Key points:

• system_prompt=agent_config.instructions — the instructions string replaces the hardcoded system prompt.
• create_strands_model(agent_config) is the provider-dispatch seam. Add a branch per provider the variation can serve.
• The tracker is Tier 3: tracker.track_duration_of(...) + an explicit track_tokens call fed by track_strands_metrics. See strands-tracking.md for the single-call track_metrics_of_async variant and the per-field breakdown of accumulated_usage.
• Always ldclient.get().flush() before process exit in short-lived scripts — trailing events can otherwise be lost.

TypeScript caveat. The Strands TypeScript SDK ships BedrockModel and OpenAIModel only — it cannot run Anthropic-backed variations. If the app needs to serve both OpenAI and Anthropic from a single config, use the Python SDK.

Custom StateGraph (bind_tools + ToolNode)

The most common LangGraph pattern in the wild is not the prebuilt agent — it's a custom StateGraph with a call_model node that does model.bind_tools(TOOLS), a separate "tools" node that runs ToolNode(TOOLS), and a conditional edge between them. This is the shape of the langchain-ai/react-agent template.

Two things make it different from the prebuilt create_agent:

1. Tools appear in two places — bind_tools(TOOLS) (so the LLM knows which tools exist) and ToolNode(TOOLS) (so the executor knows how to run them). Both must read from the same source.
2. The system prompt is injected manually in the call_model node body (usually as the first message in the ainvoke([{"role": "system", ...}, *state.messages]) call), not passed as a constructor argument.

Before — the typical template shape (src/react_agent/graph.py, src/react_agent/tools.py, src/react_agent/context.py):

# tools.py
TOOLS: List[Callable[..., Any]] = [search]

# context.py
@dataclass(kw_only=True)
class Context:
    system_prompt: str = field(default=prompts.SYSTEM_PROMPT)
    model: str = field(default="anthropic/claude-sonnet-4-5-20250929")

# graph.py
from .tools import TOOLS

async def call_model(state: State, runtime: Runtime[Context]):
    model = load_chat_model(runtime.context.model).bind_tools(TOOLS)
    system_message = runtime.context.system_prompt.format(system_time=now_iso())
    response = await model.ainvoke(
        [{"role": "system", "content": system_message}, *state.messages]
    )
    return {"messages": [response]}

builder = StateGraph(State, context_schema=Context)
builder.add_node(call_model)
builder.add_node("tools", ToolNode(TOOLS))
builder.add_edge("__start__", "call_model")
builder.add_conditional_edges("call_model", route_model_output)
builder.add_edge("tools", "call_model")
graph = builder.compile()

After — run-scoped architecture. The critical shape for the tracker factory is that one user turn = one runId = one tracker, not one LLM call = one tracker. A ReAct loop that calls call_model three times in a single turn must not mint three trackers, or billing and the Monitoring tab will treat the turn as three separate executions. The fix is to resolve the config and mint the tracker once, in a dedicated entry node, and thread both through graph state for every subsequent node.

Three nodes, in order:

1. setup_run (entry) — resolves agent_config, mints the tracker with create_tracker(), builds model with create_langchain_model(ai_config), builds tools via the factory pattern below, starts a perf_counter_ns() timer, and stashes all of it on State. Runs exactly once per turn.
2. call_model — reads model / tools / tracker / accumulator from State, runs model.ainvoke(...), accumulates token usage, calls tracker.track_tool_calls([...]) per step. Does not call track_metrics_of_async here — that wrapper records duration + success on every invocation and would fire once per iteration. On exception: call tracker.track_duration + tracker.track_error and re-raise (the finalize node will not run).
3. finalize (terminal) — runs exactly once at the end of the turn on the success path. Calls tracker.track_duration(elapsed_ms) + tracker.track_tokens(accumulated) + tracker.track_success(). Each of these fires exactly once per run, which is what the at-most-once guards enforce.

# tools.py — tool factories close over per-run config, so tools never re-fetch
from typing import Any, Callable, Dict
from langchain_core.tools import StructuredTool
from langchain_tavily import TavilySearch

def make_search(ai_config) -> Callable[..., Any]:
    """Closure: capture `max_search_results` once at setup_run time."""
    max_results = ai_config.model.get_custom("max_search_results") or 10

    async def search(query: str) -> dict:
        """Search the web for current information on a given topic."""
        return await TavilySearch(max_results=max_results).ainvoke({"query": query})

    return search

# Registry of factories keyed by LD tool name. Each factory takes the
# per-run ai_config and returns a ready-to-bind callable. This decouples
# the Config (tool metadata) from the app (implementations) and
# means tools never call get_agent_config() themselves.
TOOL_FACTORIES: Dict[str, Callable[[Any], Callable[..., Any]]] = {
    "search": make_search,
}


# graph.py
import time
from typing import Any, Dict, List, TypedDict
from uuid import uuid4
from ldai_langchain import create_langchain_model, get_ai_metrics_from_response
from ldai_langchain.langchain_helper import build_structured_tools
from ldai.tracker import TokenUsage
from ldclient import Context as LDContext
from langchain_core.runnables import RunnableConfig
from .tools import TOOL_FACTORIES


# Run-scoped State. Fields below messages[] are non-serializable and only
# meaningful inside one turn — if you add a LangGraph checkpointer, exclude
# them from the checkpoint.
class State(TypedDict, total=False):
    messages: List[Any]                 # standard LangGraph messages reducer applies
    tracker: Any                        # LDAIConfigTracker, minted in setup_run
    model: Any                          # Bound chat model with per-run tools
    tools: List[Any]                    # StructuredTool list for this turn
    start_perf_ns: int                  # time.perf_counter_ns() at setup_run
    token_accumulator: TokenUsage       # Sum of usage_metadata across loop iterations
    disabled_message: str               # Set iff ai_config.enabled is False

# If the repo's State is a @dataclass (e.g., langchain-ai/react-agent),
# attribute access works the same — `state.tools` instead of `state["tools"]`.
# Same field set, same reducers (wrap the messages field with
# `Annotated[List[Any], add_messages]`). Don't convert during migration.


async def setup_run(state: State, config: RunnableConfig) -> Dict[str, Any]:
    """Runs once per user turn. Resolves the config, mints the tracker,
    builds the model and tools, and stashes everything on state.

    Node signature takes `config: RunnableConfig` rather than
    `runtime: Runtime[Context]` so the example works with an empty Context
    dataclass and reads `thread_id` from LangGraph's standard
    `config["configurable"]["thread_id"]` plumbing. If your app has a typed
    Context schema with fields beyond thread_id, use the Runtime[Context]
    signature instead and keep those fields on Context.
    """
    configurable = config.get("configurable") or {}
    ld_key = configurable.get("thread_id") or f"anon-{uuid4()}"
    ld_context = LDContext.builder(ld_key).kind("user").build()
    ai_config = get_ai_client().agent_config(
        "react-agent",
        ld_context,
        FALLBACK,
    )

    if not ai_config.enabled:
        # Return BOTH a flag the router can short-circuit on AND an AIMessage
        # appended to state["messages"]. Downstream consumers (UI, tests, the
        # caller that invoked the graph) read the last message — setting only
        # `disabled_message` without touching `messages` produces a graph
        # whose last-message shape depends on state entry, which surprises
        # readers. Every node return should leave `messages` in a valid shape.
        return {
            "disabled_message": "Feature is currently unavailable.",
            "messages": [AIMessage(content="Feature is currently unavailable.")],
        }

    # Three-tier tool-registry contract. Don't conflate these:
    #   1. TOOL_FACTORIES          — {name: factory}   at module scope (static, never changes)
    #   2. built_callables         — {name: callable}  per-run (factories applied to ai_config)
    #   3. tools (StructuredTool)  — per-run list, ready for bind_tools + ToolNode dispatch
    # build_structured_tools reads ai_config.model.parameters.tools to decide
    # which entries from built_callables to wrap — the LLM only sees the subset
    # the variation attached, even if the registry has more callables.
    built_callables = {name: fn(ai_config) for name, fn in TOOL_FACTORIES.items()}
    tools = build_structured_tools(ai_config, built_callables)

    return {
        "tracker": ai_config.create_tracker(),
        "model": create_langchain_model(ai_config).bind_tools(tools),
        "tools": tools,
        "start_perf_ns": time.perf_counter_ns(),
        "token_accumulator": TokenUsage(input=0, output=0, total=0),
        # Cache instructions so call_model doesn't touch ai_config again.
        "instructions": ai_config.instructions or "",
    }


async def call_model(state: State) -> Dict[str, Any]:
    """Reads model / tracker / tools from state. No LD access, no config fetch."""
    tracker = state["tracker"]
    model = state["model"]
    messages = [{"role": "system", "content": state["instructions"]}, *state["messages"]]

    try:
        response = cast(AIMessage, await model.ainvoke(messages))
    except Exception:
        elapsed_ms = (time.perf_counter_ns() - state["start_perf_ns"]) // 1_000_000
        tracker.track_duration(elapsed_ms)
        tracker.track_error()
        raise

    # Accumulate token usage across loop iterations; finalize emits the sum.
    acc = state["token_accumulator"]
    if getattr(response, "usage_metadata", None):
        um = response.usage_metadata
        state["token_accumulator"] = TokenUsage(
            input=acc.input + um.get("input_tokens", 0),
            output=acc.output + um.get("output_tokens", 0),
            total=acc.total + um.get("total_tokens", 0),
        )

    if response.tool_calls:
        tracker.track_tool_calls([call["name"] for call in response.tool_calls])

    return {"messages": [response]}


async def finalize(state: State) -> Dict[str, Any]:
    """Runs exactly once at the end of a successful turn. Emits the three
    once-per-run tracker events: duration, tokens, success."""
    tracker = state["tracker"]
    elapsed_ms = (time.perf_counter_ns() - state["start_perf_ns"]) // 1_000_000
    tracker.track_duration(elapsed_ms)
    acc = state["token_accumulator"]
    if acc.total > 0:
        tracker.track_tokens(acc)
    tracker.track_success()
    return {}


def route_after_setup(state: State) -> Literal["call_model", "__end__"]:
    return "__end__" if "disabled_message" in state else "call_model"


def route_model_output(state: State) -> Literal["tools", "finalize"]:
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else "finalize"


async def tools_node(state: State) -> Dict[str, Any]:
    """Dynamic ToolNode wrapper — rebuilds dispatch per invocation.

    `ToolNode` builds its `{name: callable}` dispatch dict at construction
    time, so `ToolNode([])` cannot execute anything (dispatch returns
    "Error: <tool> is not a valid tool, try one of []." verbatim). Because
    our tool callables close over per-run `ai_config` (see TOOL_FACTORIES),
    we cannot pre-build the ToolNode at compile time — the concrete
    callables differ each turn. Construct a fresh ToolNode from
    state["tools"] per invocation.

    Two things to get right in the body:
    - Use `ainvoke`, not `invoke` — the enclosing nodes in this example
      are async and LangGraph will raise if a sync node is awaited.
    - Pass an explicit `{"messages": [...]}` payload rather than the full
      State dict — ToolNode's contract is that input shape, and passing
      extra fields has surprised users on some LangGraph versions.
    """
    return await ToolNode(list(state["tools"])).ainvoke(
        {"messages": list(state["messages"])}
    )


# No context_schema=Context here — Context can be empty or dropped entirely.
# thread_id and any other per-invocation plumbing flows through RunnableConfig
# via config["configurable"] (see setup_run). Add context_schema=Context only
# if your app truly has typed per-request context fields beyond thread_id.
builder = StateGraph(State)
builder.add_node(setup_run)
builder.add_node(call_model)
builder.add_node("tools", tools_node)  # NOT ToolNode([]) — see tools_node docstring
builder.add_node(finalize)
builder.add_edge("__start__", "setup_run")
builder.add_conditional_edges("setup_run", route_after_setup)
builder.add_conditional_edges("call_model", route_model_output)
builder.add_edge("tools", "call_model")
builder.add_edge("finalize", "__end__")
graph = builder.compile()

Why this shape:

• One create_tracker() per turn — setup_run is the only caller. Each factory call mints a fresh runId; per-iteration calls fragment the run downstream and reset the at-most-once guards.
• One agent_config(...) per turn — setup_run resolves once. Re-fetching in a loop step inflates agent-config event counts and lets a mid-turn targeting change swap variations between LLM calls.
• track_duration / track_tokens / track_success fire in finalize, not call_model — these are at-most-once; per-iteration calls are silently dropped.
• track_tool_calls in call_model is fine — it's per-event metadata, not at-most-once.
• Tools close over per-run config — make_search(ai_config) captures max_search_results at setup time. A mid-turn variation change doesn't affect the in-flight turn; the next turn picks up the new value.
• tools node uses the tools_node wrapper, not ToolNode([]) — ToolNode builds its dispatch dict at construction, so an empty list produces empty dispatch and every tool call fails with is not a valid tool. The wrapper rebuilds per invocation from state.
• Fallback instructions use Mustache ("...{{ system_time }}"), not .format() — the SDK's Mustache renderer runs on both the LD-served path and the fallback path; single-brace placeholders ship a stale literal.

Gotchas:

• Never call track_metrics_of_async in a loop node. trackMetricsOf is designed for single-call shapes (one provider call, one success event). In a ReAct loop it would re-fire track_success per iteration and trip the at-most-once guard. Use manual track_tool_calls per step and explicit track_duration + track_tokens + track_success in finalize.
• Lazy-init the ai_client. Avoid ai_client = LDAIClient(...) at module import time — it couples test collection to LD initialization and makes the per-turn runId story harder to mock. Wrap with def get_ai_client(): ... and cache on first use.
• Per-run LDContext keys — and their MAU cost. A shared literal like "anonymous" collapses every request into one targeting/billing segment, which breaks experimentation and per-run segmentation. The obvious fix is to use the LangGraph thread_id if present and fall back to uuid4() per setup_run. Before doing this in production, check the MAU impact: LaunchDarkly bills by distinct context keys per month, so a production agent serving 100k anonymous runs/day will register 3M+ MAU even though there's no real user. Three shapes, pick the one that fits:
  1. Known user identity (preferred). If the caller knows who the user is (session cookie, auth token, SSO ID), use that as the context key. No anonymous keys at all. Targeting, segmentation, and MAU all work correctly.
  2. Session-scoped key. Use the LangGraph thread_id, or the chat-session ID, or whatever the longest-lived identifier is below "real user." MAU scales with session count, not turn count.
  3. Per-turn uuid4(). Only in demos, or if you genuinely need per-run isolation for experiment targeting and are willing to pay the MAU. Document the decision in the repo README so the next migrator doesn't swap it out without reading the tradeoff.
• State field serialization. If you add a LangGraph checkpointer, exclude the run-scoped fields (tracker, model, tools, start_perf_ns) from the checkpoint — they're non-serializable and only meaningful inside one turn anyway.

Node.js — LangGraph.js createReactAgent (prebuilt)

For apps built on @langchain/langgraph's prebuilt createReactAgent, the loop happens inside one agent.invoke(...) call — you don't own the nodes. That makes the run-scoped pattern simpler than the Python custom-StateGraph shape above: resolve agentConfig once, mint the tracker once, wrap the whole agent.invoke in one tracker.trackMetricsOf(...) call per user turn.

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

const ldClient = init(process.env.LD_SDK_KEY!);
await ldClient.waitForInitialization({ timeout: 10 });
const aiClient = initAi(ldClient);

// Sum token usage across every message the agent produced in one turn.
// Multiple fallback field names cover the provider variation LangChain
// normalizes over (OpenAI, Anthropic, Bedrock, Gemini, …).
function langgraphMetrics(result: any): LDAIMetrics {
  let input = 0, output = 0, total = 0;
  for (const msg of result.messages ?? []) {
    const usage = msg.response_metadata?.token_usage ?? msg.usage_metadata;
    if (!usage) continue;
    input += usage.input_tokens ?? usage.prompt_tokens ?? usage.promptTokens ?? 0;
    output += usage.output_tokens ?? usage.completion_tokens ?? usage.completionTokens ?? 0;
    total += usage.total_tokens ?? usage.totalTokens ?? 0;
  }
  if (total === 0) total = input + output;
  return { success: true, tokens: total > 0 ? { input, output, total } : undefined };
}

async function runTurn(userInput: string, threadId: string): Promise<string | null> {
  const context = { kind: 'user' as const, key: threadId };
  const agentConfig: LDAIAgentConfig = await aiClient.agentConfig(
    'react-agent',
    context,
    FALLBACK,                            // LDAIAgentConfigDefault literal
  );

  if (!agentConfig.enabled) return null;

  // Build everything once per user turn.
  const llm = await createLangChainModel(agentConfig);
  const agent = createReactAgent({
    llm,
    tools: buildTools(agentConfig),      // factory pattern — see below
    prompt: agentConfig.instructions,
    checkpointer: new MemorySaver(),     // reuse across turns if you want chat memory
  });

  // One tracker per user turn. Fresh runId. At-most-once guards reset.
  const tracker = agentConfig.createTracker();

  // Exceptions are tracked automatically — trackMetricsOf catches
  // exceptions, records tracker.trackError(), and re-throws.
  const result = await tracker.trackMetricsOf(
    langgraphMetrics,
    () => agent.invoke(
      { messages: [{ role: 'user', content: userInput }] },
      { configurable: { thread_id: threadId } },
    ),
  );
  const messages = result.messages ?? [];
  return messages.length ? String(messages[messages.length - 1].content) : null;
}

Tool factories in TypeScript. The same pattern as Python — a record of (agentConfig) => tool factories, applied per-turn so each tool closes over the live variation's model.custom knobs:

import { tool } from '@langchain/core/tools';
import { z } from 'zod';

type ToolFactory = (agentConfig: LDAIAgentConfig) => ReturnType<typeof tool>;

function makeSearch(agentConfig: LDAIAgentConfig) {
  const maxResults =
    (agentConfig.model?.custom?.max_search_results as number | undefined) ?? 10;
  return tool(
    async ({ query }: { query: string }) => {
      const res = await fetch(`https://api.tavily.com/search?q=${encodeURIComponent(query)}&n=${maxResults}`);
      return await res.json();
    },
    {
      name: 'search',
      description: 'Search the web for current information on a given topic.',
      schema: z.object({ query: z.string() }),
    },
  );
}

const TOOL_FACTORIES: Record<string, ToolFactory> = {
  search: makeSearch,
};

function buildTools(agentConfig: LDAIAgentConfig) {
  const attached = ((agentConfig.model?.parameters?.tools as Array<{ name: string }>) ?? [])
    .map((t) => t.name);
  return attached
    .filter((name) => name in TOOL_FACTORIES)
    .map((name) => TOOL_FACTORIES[name](agentConfig));
}

Key differences from Python custom-StateGraph:

• Because LangGraph.js's createReactAgent is prebuilt, you don't write a setup_run / call_model / finalize graph — agent.invoke(...) is the single call that represents the turn, so wrapping it in one trackMetricsOf call is sufficient. The at-most-once guards are naturally satisfied by the single wrapping call.
• No dynamic ToolNode wrapper needed. createReactAgent takes a tool list at construction; construct the agent per-turn (inside runTurn) so the tools can close over the current agentConfig.
• Multi-turn chat: if you want runTurn to be called three times in a row for one session (shared threadId / MemorySaver), each call still mints its own tracker inside runTurn. The threadId threads conversation memory through checkpointer; the runId identifies each turn independently.

Custom StateGraph in Node.js. If the app uses a hand-rolled StateGraph with its own call_model / tool nodes (uncommon in Node compared to Python, but possible), the same run-scoped architecture from the Python section above applies — setup_run as an entry node, tools_node wrapper around new ToolNode(state.tools).invoke(...), finalize as a terminal node. The TypeScript syntax differs but the node responsibilities are identical. Prefer createReactAgent unless the app genuinely needs graph-level control.

Custom ReAct loop

The same run-scoped shape applies: one config fetch + one create_tracker() at the top of the turn, accumulate tokens across the loop, emit track_duration / track_tokens / track_success exactly once after the loop exits.

import time
from ldai.tracker import TokenUsage

def run_turn(ai_client, user_id: str, user_input: str):
    context = Context.builder(user_id).kind("user").build()
    config = ai_client.agent_config("custom-react", context, FALLBACK)

    if not config.enabled:
        return disabled_response()

    # Resolve everything needed for the turn once, up top.
    system_prompt = config.instructions
    model_name = config.model.name
    tracker = config.create_tracker()
    tool_callables = build_tools_from_config(config)  # closes over config
    start_ns = time.perf_counter_ns()
    acc = TokenUsage(input=0, output=0, total=0)

    try:
        history = [{"role": "user", "content": user_input}]
        for step in range(MAX_TURNS):
            response = my_provider.complete(
                model=model_name,
                system=system_prompt,
                messages=history,
                tools=config.model.get_parameter("tools") or [],
            )

            # Accumulate token usage across the loop; finalize emits the sum.
            usage = extract_tokens(response)  # returns TokenUsage
            acc = TokenUsage(
                input=acc.input + usage.input,
                output=acc.output + usage.output,
                total=acc.total + usage.total,
            )

            tool_calls = extract_tool_calls(response)
            if tool_calls:
                tracker.track_tool_calls([c["name"] for c in tool_calls])
                history.extend(run_tools(tool_calls, tool_callables))
                continue

            # Done — fall through to the success emit block.
            break

        elapsed_ms = (time.perf_counter_ns() - start_ns) // 1_000_000
        tracker.track_duration(elapsed_ms)
        if acc.total > 0:
            tracker.track_tokens(acc)
        tracker.track_success()
        return response
    except Exception:
        elapsed_ms = (time.perf_counter_ns() - start_ns) // 1_000_000
        tracker.track_duration(elapsed_ms)
        tracker.track_error()
        raise

The call site stays in your control; the config just delivers instructions, model.name, model.parameters, and tools. Everything that's stable across the turn (model name, instructions, tool bindings, tracker) is hoisted out of the loop body — the loop itself only does message passing and tool dispatch.

Do not call track_duration / track_tokens / track_success inside the for body. The at-most-once guards will warn and drop the second-and-later calls on the same tracker, so per-step tracker calls will silently lose data. Accumulate inside the loop, emit once after.

Dynamic tool loading — the "tools factory" pattern

The devrel-agents-tutorial uses a dynamic tool factory that reads tool names from config.tools and instantiates the actual tool implementations at runtime. This decouples the config (which holds tool metadata) from the application (which holds the executable code).

The pattern

# tools_impl/dynamic_tool_factory.py — adapted from devrel-agents-tutorial

def extract_tool_names(config) -> list[str]:
    """Read the list of tool names from the config."""
    if not hasattr(config, "tools") or not config.tools:
        return []
    return [tool.name if hasattr(tool, "name") else tool.get("name") for tool in config.tools]


def create_dynamic_tools_from_launchdarkly(config) -> list:
    """Instantiate tool implementations for every tool name on the config."""
    tool_names = extract_tool_names(config)
    instances = []
    for name in tool_names:
        tool = _create_tool_instance(name)
        if tool is not None:
            instances.append(tool)
    return instances


def _create_tool_instance(tool_name: str):
    """Map a tool name to an actual implementation. Add one branch per tool."""
    if tool_name == "search_kb":
        from my_tools.search import SearchKBTool
        return SearchKBTool()
    elif tool_name == "calculator":
        from my_tools.calc import CalculatorTool
        return CalculatorTool()
    # ... etc
    return None

Then at the call site:

config = ai_client.agent_config("support-agent", context, FALLBACK)
tools = create_dynamic_tools_from_launchdarkly(config)

agent = create_agent(
    build_llm(config),
    tools,
    system_prompt=config.instructions,
)

What this gives you:

• Toggle a tool on/off by editing the config in LaunchDarkly — no redeploy needed to remove a tool from production
• Roll out a new tool to 5% of users by editing targeting rules (combined with configs-targeting)
• Keep the actual tool implementation code in the repo; only metadata lives in LaunchDarkly

Extracting schemas from existing hardcoded tools

If your tools are defined with LangChain @tool or Pydantic BaseModel schemas, extract the JSON schema programmatically to pass to tools during Stage 3:

from my_tools import search_kb  # a @tool-decorated function

schema = search_kb.args_schema.model_json_schema()
# schema is a dict ready to pass as the tool's parameters field

Do not hand-write the schema — LangChain already generated it from the function signature, and Pydantic will keep it in sync. The tools delegate accepts raw JSON Schema for the parameters field.

Dynamic schemas from LaunchDarkly

The devrel tutorial also shows the reverse: reading a JSON schema from the config and constructing a Pydantic model at runtime:

from pydantic import BaseModel, Field, create_model

def _create_dynamic_tool_input(tool_config: dict) -> type[BaseModel]:
    """Build a Pydantic input schema from a config tool's parameters."""
    properties = tool_config.get("properties", {})
    fields = {}
    for name, cfg in properties.items():
        py_type = {"string": str, "number": float, "integer": int, "boolean": bool}.get(
            cfg.get("type"), str
        )
        default = ... if name in tool_config.get("required", []) else None
        fields[name] = (py_type, Field(default=default, description=cfg.get("description", "")))
    return create_model("DynamicToolInput", **fields)

This lets LaunchDarkly change a tool's parameter schema without redeploying the app. Use it when the tool implementation is generic enough to accept any parameter shape (e.g. a proxy that forwards requests to an external API). For most tools, a static Pydantic schema in the repo is simpler.

Routing / multi-node hint

If the framework needs to pick between multiple downstream agents (e.g. a supervisor that routes to a security agent or a support agent based on the user input), do not roll your own routing. Use LaunchDarkly agent graphs — the graph's edges carry the routing contract, and the supervisor's instructions can be auto-injected with the valid routes.

The devrel tutorial's generic_agent.py shows a minimal version of this:

# If this node has outgoing edges with routes, inject them into instructions
if self.valid_routes:
    route_instruction = (
        f"\n\nYou must select one of these routes: {self.valid_routes}. "
        f'Return your choice in JSON format: {{"route": "<selected_route>"}}'
    )
    instructions = instructions + route_instruction

For the full graph pattern, read agent-graph-reference.md — but again, single-agent migration comes first, and agent graphs are currently Python-only in the SDK.

Keep the provider call in the repo

One rule that applies across all three frameworks: the provider SDK call (OpenAI, Anthropic, Bedrock) stays in your code. The config only changes the inputs to that call — model name, instructions, parameters, tool list. It does not replace the provider SDK. That means:

• You keep full control of error handling, retries, timeouts, custom headers
• You keep full control of streaming logic and backpressure
• You keep full control of authentication (API keys, IAM roles, Bedrock Converse sessions)
• The config is additive — removing it gets you back to the original hardcoded app, provided the fallback mirrors the old values