Getting started with Strands and AgentControl configs

import asyncio
import os
import ldclient
from ldclient import Context
from ldclient.config import Config
from ldai import LDAIClient
from ldai.tracker import TokenUsage
from ldai.providers import LDAIMetrics
from strands import Agent, tool
from strands.models.anthropic import AnthropicModel
from strands.models.openai import OpenAIModel
from strands.models.bedrock import BedrockModel
from strands.agent.conversation_manager.sliding_window_conversation_manager import (
    SlidingWindowConversationManager,
)
from dotenv import load_dotenv

load_dotenv()

SDK_KEY = os.environ.get("LAUNCHDARKLY_SDK_KEY")
AGENT_CONFIG_KEY = "strands-agent"

@tool
def get_order_status(order_id: str) -> str:
    """Look up the status of a customer order by order ID."""
    orders = {
        "ORD-123": "Shipped — arrives Thursday",
        "ORD-456": "Processing — estimated ship date: tomorrow",
        "ORD-789": "Delivered on Monday",
    }
    return orders.get(order_id, f"No order found with ID {order_id}")


# Map tool keys (matching the LaunchDarkly tool keys) to local handlers. The agent
# build step resolves the active tool list from `agent_config.tools` at runtime.
TOOL_REGISTRY = {"get_order_status": get_order_status}


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 {})
    # LaunchDarkly surfaces attached tools from `parameters.tools` in its own flat shape.
    # Drop the key here. Strands receives tools from the agent constructor.
    params.pop("tools", None)

    is_bedrock = provider == "bedrock" or model_id.startswith(
        ("us.", "eu.", "apac.", "anthropic.", "amazon.", "meta.")
    )

    if is_bedrock:
        region = (
            params.pop("region_name", None)
            or os.environ.get("AWS_REGION")
            or "us-west-2"
        )
        known = {
            k: params.pop(k)
            for k in ("max_tokens", "temperature", "top_p", "stop_sequences")
            if k in params
        }
        if "max_tokens" not in known:
            known["max_tokens"] = 1024
        return BedrockModel(
            model_id=model_id,
            region_name=region,
            additional_request_fields=params or None,
            **known,
        )
    if provider == "anthropic":
        # AnthropicModel requires max_tokens as a kwarg, not in 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 unchanged. GPT-5 wants `max_completion_tokens`,
        # GPT-4o wants `max_tokens`. Keep that choice in the LaunchDarkly variation.
        return OpenAIModel(model_id=model_id, params=params)
    raise ValueError(f"Unsupported provider for Strands: {provider!r}")


def strands_metrics_extractor(result):
    """Pull token usage and tool calls off a Strands AgentResult into an 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)

    tool_calls = []
    for tool_name, tm in (result.metrics.tool_metrics or {}).items():
        tool_calls.extend([tool_name] * tm.call_count)

    return LDAIMetrics(
        success=True,
        tokens=TokenUsage(input=input_tokens, output=output_tokens, total=total) if total > 0 else None,
        tool_calls=tool_calls or None,
    )


async def async_main():
    ldclient.set_config(Config(SDK_KEY))
    if not ldclient.get().is_initialized():
        print("LaunchDarkly SDK failed to initialize")
        return

    ai_client = LDAIClient(ldclient.get())

    context = Context.builder("user-123").kind("user").name("Sandy").build()

    # Pass a default for improved resiliency when the AgentControl config is unavailable
    # or LaunchDarkly is unreachable. Omit it to disable the default.
    # Example:
    #   from ldai import AIAgentConfigDefault
    #   default = AIAgentConfigDefault(
    #       enabled=True,
    #       model={"name": "gpt-5"},
    #       provider={"name": "openai"},
    #       instructions="You are a helpful assistant.",
    #   )
    #   agent_config = ai_client.agent_config(AGENT_CONFIG_KEY, context, default)
    agent_config = ai_client.agent_config(AGENT_CONFIG_KEY, context)

    if not agent_config.enabled:
        print("Agent Config is disabled — run the notebook to set up the config")
        return

    model = create_strands_model(agent_config)

    # Resolve the agent's tool list from the LaunchDarkly variation. AIAgentConfig.tools
    # is the typed surface in SDK 0.20+ (a dict of name -> LDTool) — no more digging
    # through model.parameters.
    ld_tool_names = list(agent_config.tools or {})
    resolved_tools = [TOOL_REGISTRY[n] for n in ld_tool_names if n in TOOL_REGISTRY]

    # SlidingWindowConversationManager gives the agent short-term memory across turns.
    conversation_manager = SlidingWindowConversationManager(window_size=40)

    agent = Agent(
        name="order-assistant",
        model=model,
        system_prompt=agent_config.instructions,
        tools=resolved_tools,
        conversation_manager=conversation_manager,
    )

    async def run_turn(agent, user_input):
        # Fresh tracker per turn (each is one execution: own runId, at-most-once
        # for success/error/tokens/duration/tool_call). The extractor flattens
        # `result.metrics.tool_metrics` into the `tool_calls` field of LDAIMetrics,
        # which the SDK turns into one track_tool_call event per invocation.
        tracker = agent_config.create_tracker()
        try:
            result = await tracker.track_metrics_of_async(
                strands_metrics_extractor,
                lambda: agent.invoke_async(user_input),
            )
            print(f"Agent: {result.message['content'][0]['text']}")
        except Exception as e:
            print(f"Error: {e}")

    await run_turn(agent, "What's the status of order ORD-123?")
    await run_turn(agent, "What about ORD-456?")
    await run_turn(agent, "Summarize both orders for me.")

    # Always flush events before closing. Otherwise, trailing events are at risk of being
    # lost, in both short-lived scripts and long-running services.
    ldclient.get().flush()
    ldclient.get().close()


def main():
    asyncio.run(async_main())


if __name__ == "__main__":
    main()