launchdarkly-ai-server — Core Client

The core package for the LaunchDarkly AI Python SDK. It owns the LaunchDarkly client lifecycle, telemetry pipeline, all shared types, and the primary entry points that handler packages depend on.

All handler packages (launchdarkly-ai-*) depend on this package.

Tip: for the simplest install, use launchdarkly-ai instead. It re-exports this package's full API and is the recommended default for most applications.

Installation

Without telemetry

pip install launchdarkly-ai-server

launchdarkly-server-sdk is an optional dependency — include it for standard usage, or pass a pre-initialized client to init_client(client=...) if you bring your own.

The SDK works fully without the OpenTelemetry packages — feature flags evaluate, handlers run, and LaunchDarkly AI events are tracked. Spans are created as no-ops. If you call init_client() without the OTel packages installed, the SDK logs a single warning and continues normally.

With telemetry (recommended for production)

To export traces to the LaunchDarkly Observability dashboard (or any OTLP-compatible backend), install the otel extras group:

pip install "launchdarkly-ai-server[otel]"

No code changes are required — init_client() detects the packages at runtime and sets up the tracer provider automatically.

Environment Variables

Variable	Required	Description
LD_SDK_KEY	Yes	LaunchDarkly server-side SDK key
LD_BASE_URI	No	Override the LaunchDarkly polling base URI (e.g. for staging)
LD_STREAM_URI	No	Override the streaming URI
LD_EVENTS_URI	No	Override the events URI
LD_SERVICE_NAME	No	OTel service.name resource attribute (default: python-sdk)
LD_ENVIRONMENT	No	deployment.environment resource attribute attached to telemetry
OTEL_EXPORTER_OTLP_ENDPOINT	No	OTLP endpoint override (default: LaunchDarkly Observability backend)
LD_API_TOKEN	For evaluations	API access token used by the evaluations management API
LD_SDK_KEY	For evaluations	SDK key whose event transport carries generation results to LaunchDarkly
LD_API_BASE_URI	No	Evaluations management API host override; intentionally separate from LD_BASE_URI
LD_UI_BASE_URI	No	LaunchDarkly application host for evaluation-run links (default: https://app.launchdarkly.com; staging: https://ld-stg.launchdarkly.com)

Run an evaluation from code

The generation-only evaluations harness reads an LD-hosted dataset, creates a new evaluation and API-source run, and invokes your handler once per row. Each success or error queues a $ld:ai:offline-evals:generation custom event containing the evaluation, run, dataset, and row identifiers plus output or error (errorMessage is included for ERROR rows), nested usage.inputTokens/usage.outputTokens, timing, and stable hashes. Dataset-owned input, expected output, metadata, and variables are not duplicated in the event. Each queued event prints a line to stdout with its RFC3339 UTC emittedAt timestamp and stable eventId, making it possible to compare SDK emission time with ClickHouse arrival time. The same emittedAt value is included in the event payload. Events are flushed before the summary is fetched and the call returns; handlers are never rerun to retry event delivery. Pass/fail is derived from LaunchDarkly's run summary.

Result links use ui_base_uri, then LD_UI_BASE_URI, then https://app.launchdarkly.com; this is independent of LD_API_BASE_URI. After flushing generation events, the harness polls the run summary endpoint until passed + failed + error rows fully account for a nonzero total with no pending rows, polling every poll_interval_seconds (default 2s) up to poll_timeout_seconds (default 180s); pass either to run() to widen both for large datasets. The summary endpoint does not return run state, so RunSummary exposes row counts only. A generation result passes only when the completed summary has no error or pending rows. Evaluation keys must be unique because every call creates a new evaluation with POST.

import asyncio
import sys

from launchdarkly_ai_openai_messages import create_openai_messages_handler
from launchdarkly_ai_server import init_evaluations


async def main() -> int:
    evals = init_evaluations()  # LD_API_TOKEN required; LD_SDK_KEY unless a client is already initialized
    result = await evals.run(
        project_key="my-project",
        key="support-qa-2026-08-20",
        dataset="support-golden",
        handler=create_openai_messages_handler(),
        generation={
            "provider": "OpenAI",
            "model": "gpt-4o",
            "instructions": "You are a support agent.",
        },
    )
    print(result.url, result.summary)
    return 0 if result.passed else 1


sys.exit(asyncio.run(main()))

project_key is supplied per run rather than during initialization. generation.instructions is shorthand for one system message; use generation.messages instead for a full message list, but do not supply both. The harness never retries a handler invocation because doing so could repeat tool side effects. Its retries apply only to LaunchDarkly management API requests.

Generation events are the only path by which row results reach LaunchDarkly, so init_evaluations() raises rather than creating a run that can never complete unless it can resolve an event transport: either an SDK key (sdk_key or LD_SDK_KEY) or a client already initialized through init_client(client=...). Bringing your own client lets a process emit evaluation events without an SDK key in scope. Every generated row is emitted and flushed unconditionally; no feature flag gates event publishing. The harness then polls the summary endpoint until row accounting shows processing is complete.

The client uses lazy initialization: importing the package does not connect to LaunchDarkly. The singleton is created automatically on the first API call that needs it (config().invoke(), graph().invoke(), resolve_graph(), etc.), as long as LD_SDK_KEY is set in the environment.

Call init_client() explicitly when you want to:

• Pass SDK or telemetry options programmatically (overriding env vars)
• Initialize at startup before the first AI call (e.g. to avoid latency on the first request)
• Fail fast at boot if LD_SDK_KEY is missing

import asyncio
from launchdarkly_ai_server import init_client, shutdown

async def main():
    # Standard path — auto-discovers launchdarkly-server-sdk.
    client = await init_client({
        "sdkKey": "sdk-...",
        "serviceName": "my-service",
        "environment": "production",
    })

    # Or skip init_client() and let the first model/graph call initialize lazily.

    # Flush telemetry, flush LD events, and close the client.
    await shutdown()

asyncio.run(main())

Export	Description
init_client(options?)	Auto-discover and initialize launchdarkly-server-sdk. Optional — the first AI API call triggers lazy init when LD_SDK_KEY is set. Returns Awaitable[LDClientInterface].
init_client(client=...)	BYOC overload — accept a pre-initialized LDClientInterface. Skips SDK auto-discovery.
get_client()	Return the initialized LDClientInterface. Raises if init_client has not completed.
shutdown()	Flush all events and telemetry, then close the client. Await before process exit.
inspect_config(key, context)	Read an AI Config variation without invoking the model. Never raises. Returns {"enabled", "config", "meta"}.

config(**args)

The primary entry point for AI config invocations. Accepts either a single handler or a list of handlers and routes to the correct one at invoke-time based on the flag variation's provider and mode.

import asyncio
from launchdarkly_ai_server import config, shutdown
from launchdarkly_ai_openai_messages import create_openai_messages_handler
from launchdarkly_ai_openai_agents import create_openai_agent_handler
from launchdarkly_ai_claude_agents import create_claude_agents_handler

# Single handler — must match the flag variation's provider+mode, or raises.
caller = config(
    key="my-ai-config-flag",
    handler=create_openai_messages_handler(),
    tool_handlers={"my_tool": my_tool_fn},  # optional: tool implementations
)

async def main():
    result = await caller.invoke(
        "What is feature flagging?",
        {"kind": "user", "key": "user-123"},
        {"user_name": "Alice"},             # optional: template substitutions
    )
    print(result.response)  # str
    print(result.usage)     # {"input": ..., "output": ..., "total": ...}

    # Multiple handlers — routing selects the match by provider + mode.
    router = config(
        key="my-ai-config-flag",
        tool_handlers={"search": search_fn},
        handler=[
            create_openai_messages_handler(),  # provides_for: ["OpenAI", "messages"]
            create_openai_agent_handler(),     # provides_for: ["OpenAI", "agent"]
            create_claude_agents_handler(),    # provides_for: ["Anthropic", "agent"]
        ],
    )
    result2 = await router.invoke("Summarize this document", {"kind": "user", "key": "user-123"})
    print(result2.judge_results)  # judge evaluation results when skip_judges=False (default)
    print(result2.track_data)     # run ID, config key, model name, etc.

    # Multi-turn conversation — pass prior turns as history (4th arg after variables).
    history = [
        {"role": "user", "content": "What is feature flagging?"},
        {"role": "assistant", "content": "Feature flagging is a technique for safely releasing features..."},
    ]
    result3 = await caller.invoke("Can you give me an example?", {"kind": "user", "key": "user-123"}, None, history)
    await shutdown()

asyncio.run(main())

graph(key, **options)

Runs a multi-agent workflow defined in a LaunchDarkly agent graph flag. The SDK uses a model-driven router: it starts at the root node, presents outgoing edges as handoff choices to the model, and follows whichever edge the model selects. The loop terminates when the model produces a final answer, a leaf is reached, a cycle is detected, or the step cap is hit.

Each node runs through the same tracked path as config().invoke(), so every node emits its own telemetry and judges. Graph-level $ld:ai:graph:* events wrap the full run.

import asyncio
from launchdarkly_ai_server import graph, shutdown
from launchdarkly_ai_claude_agents import create_claude_agents_handler

async def main():
    g = graph(
        "support-graph",
        handlers=[create_claude_agents_handler()],
        tool_handlers={"search": search_fn},
    )

    result = await g.invoke(
        "I was double charged",
        {"kind": "user", "key": "user-123"},
        {"account_tier": "pro"},  # optional variables
    )

    print(result.response)  # final output
    print(result.usage)     # UsageDict with .input, .output, .total
    await shutdown()

asyncio.run(main())

resolve_graph(key, *, context, **options) returns a GraphDefinition without executing it. The definition carries enabled so you can branch on a disabled graph before traversing. graph(...).invoke() raises if the graph is disabled.

Registry / global_registry / compose

A Registry bundles handlers and tool implementations that can be shared across config(), graph(), and resolve_graph() calls. Pass it as registry=...; local handler/tool_handlers always take precedence.

from launchdarkly_ai_server import Registry, global_registry, compose, config
from launchdarkly_ai_claude_agents import create_claude_agents_handler

# Build a reusable registry
my_registry = Registry(
    handlers=[create_claude_agents_handler()],
    tools={"my_tool": my_tool_fn},
)

# Or register incrementally
my_registry.register(tools={"another_tool": another_fn})

# Use global_registry as a process-wide default
global_registry.register(handlers=[create_claude_agents_handler()])

# Combine two registries — b wins over a on conflict, neither is mutated
combined = compose(my_registry, another_registry)

router = config(key="my-flag", registry=my_registry)

inspect_config(key, context)

Reads an AI Config flag variation without invoking any AI provider. Use this for health checks, logging, feature-gate probes, or any situation where you need to know whether a config is enabled or what model it points to — without spending API quota.

import asyncio
from launchdarkly_ai_server import inspect_config

async def main():
    result = await inspect_config("my-ai-config-flag", {"kind": "user", "key": "user-123"})

    if not result["enabled"]:
        print("Flag is off — skipping AI call")
    else:
        print(result["config"]["model"]["name"])  # e.g. "claude-opus-4-5"
        print(result["meta"]["variationKey"])

asyncio.run(main())

Guarantees:

• Never raises — returns {"enabled": False, "config": None, "meta": None} on any error (network failure, bad key, schema mismatch, etc.)
• Does not emit LD telemetry events
• Does not call any AI provider
• Lazily initializes the LD client (same as all other entry points)

Return key	Type	Description
enabled	bool	Whether the flag variation is active
config	dict | None	The parsed AI config, or None when disabled or invalid
meta	dict | None	Variation metadata (key, version, mode), or None when unreachable

---

Utility Helpers

from launchdarkly_ai_server import parse_template, parse_json_with_possible_fences

# Replaces {{variable}} placeholders, supports dot-notation ({{user.name}})
prompt = parse_template("Hello, {{name}}!", {"name": "Alice"})

# Parses JSON that may be wrapped in ```json fences
data = parse_json_with_possible_fences(model_output)

Shared Types

All types are exported from this package. Handler packages import them from here and never redefine them.

Type	Description
AiConfigRep	The AI configuration object fetched from a LaunchDarkly flag variation
Tool	A tool definition (name, description, JSON Schema parameters)
ProviderHandler	The callable type that all handler packages produce
ProviderResponse	The value returned to callers: response, usage, track_data, judge_results?, judge_tasks?. judge_results is populated when skip_judges=False; judge_tasks (a list[JudgeTask]) is populated when skip_judges=True.
ConfigArgs	Arguments accepted by config() (key, handler, tool_handlers, registry)
NativeTool	Marker class for provider built-in tools
LDContext	Standard LaunchDarkly context dict. Import from launchdarkly_ai_server.
GraphOptions	Options accepted by graph() (handlers, tool_handlers, graph_judge — no context)
GraphDefinition	A resolved agent graph: topology accessors, run_node, and the traverse primitives (attribute access, e.g. gd.enabled, gd.get_node(key))
GraphNode / GraphEdge	A dataclass node (.key, .config, .meta, .edges, .is_terminal) and a dataclass directed edge (.key, .source_key, .target_key, .handoff)
ProviderGraphResponse	A dataclass returned by graph(...).invoke(): .response, .usage, .judge_results
GraphTopology	The parsed graph flag shape (root + edges)