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)

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)