Getting started with Amazon Bedrock and AI Configs

$
pip install "launchdarkly-server-sdk-ai>=0.17.0"
$
pip install "boto3>=1.34.0"
$
pip install "python-dotenv>=1.0.0"

1
import os
2
import boto3
3
import ldclient
4
from ldclient import Context
5
from ldclient.config import Config
6
from ldai.client import LDAIClient, AICompletionConfigDefault, AIAgentConfigDefault
7
from dotenv import load_dotenv
8
9
load_dotenv()
10
11
SDK_KEY = os.environ.get("LAUNCHDARKLY_SDK_KEY")
12
CONFIG_KEY = "bedrock-assistant"
13
AGENT_CONFIG_KEY = "bedrock-agent"
14
15
bedrock_client = boto3.client(
16
    'bedrock-runtime',
17
    region_name=os.environ.get('AWS_REGION', 'us-west-2'),
18
)
19
ldclient.set_config(Config(SDK_KEY))
20
if not ldclient.get().is_initialized():
21
    exit(1)
22
23
ai_client = LDAIClient(ldclient.get())

1
# Define the context for the current user
2
context = Context.builder("user-123") \
3
    .kind("user") \
4
    .name("Sandy") \
5
    .build()
6
7
# Pass a default for improved resiliency when the AI config is unavailable
8
# or LaunchDarkly is unreachable; omit for a disabled default.
9
# Example:
10
#   from ldai.client import AICompletionConfigDefault
11
#   default = AICompletionConfigDefault(
12
#       enabled=True,
13
#       model={"name": "us.anthropic.claude-sonnet-4-20250514-v1:0"},
14
#       provider={"name": "bedrock"},
15
#       messages=[{"role": "system", "content": "You are a helpful assistant."}],
16
#   )
17
#   config = ai_client.completion_config(CONFIG_KEY, context, default, variables={"topic": "Python"})
18
19
# Get the AI Config
20
config = ai_client.completion_config(CONFIG_KEY, context, variables={"topic": "Python"})
21
tracker = config.tracker

1
def inference_config_from_params(params):
2
    """Map AI Config parameters to Bedrock's Converse inferenceConfig shape.
3
    Rename max_tokens to maxTokens; keep temperature, topP, stopSequences as-is."""
4
    mapping = {"max_tokens": "maxTokens"}
5
    allowed = {"maxTokens", "temperature", "topP", "stopSequences"}
6
    renamed = {mapping.get(k, k): v for k, v in (params or {}).items()}
7
    return {k: v for k, v in renamed.items() if k in allowed}

1
if config.enabled:
2
    tracker = config.tracker
3
    messages = config.messages or []
4
5
    # Bedrock Converse: system messages are a top-level list, not a role
6
    system_messages = [{'text': m.content} for m in messages if m.role == 'system']
7
    conversation = [
8
        {'role': m.role, 'content': [{'text': m.content}]}
9
        for m in messages if m.role in ('user', 'assistant')
10
    ]
11
    conversation.append({
12
        'role': 'user',
13
        'content': [{'text': 'How do I read a file in Python?'}]
14
    })
15
16
    converse_params = {
17
        'modelId': config.model.name,
18
        'messages': conversation,
19
    }
20
    if system_messages:
21
        converse_params['system'] = system_messages
22
    ld_params = (config.model.to_dict().get("parameters") if config.model else None) or {}
23
    inference = inference_config_from_params(ld_params)
24
    if inference:
25
        converse_params['inferenceConfig'] = inference
26
27
    response = tracker.track_bedrock_converse_metrics(
28
        bedrock_client.converse(**converse_params)
29
    )

1
agent = ai_client.agent_config(
2
    AGENT_CONFIG_KEY,
3
    context,
4
    AIAgentConfigDefault(enabled=False),
5
)
6
7
if agent.enabled:
8
    tracker = agent.tracker
9
    ld_params = (agent.model.to_dict().get("parameters") if agent.model else None) or {}
10
    inference = inference_config_from_params(ld_params)
11
12
    # LD returns attached tools under parameters.tools in OpenAI type=function shape.
13
    # Bedrock Converse wants {toolSpec: {name, description, inputSchema: {json}}}.
14
    ld_tools = ld_params.get("tools", []) or []
15
    tool_config = {
16
        "tools": [
17
            {
18
                "toolSpec": {
19
                    "name": t["name"],
20
                    "description": t.get("description", ""),
21
                    "inputSchema": {"json": t.get("parameters", {"type": "object", "properties": {}})},
22
                }
23
            }
24
            for t in ld_tools
25
        ]
26
    }
27
28
    # Handlers stay in application code — LD governs the schema, the app owns execution.
29
    def get_order_status(order_id: str) -> str:
30
        orders = {
31
            "ORD-123": "Shipped — arrives Thursday",
32
            "ORD-456": "Processing — estimated ship date: tomorrow",
33
            "ORD-789": "Delivered on Monday",
34
        }
35
        return orders.get(order_id, f"No order found with ID {order_id}")
36
37
    tool_handlers = {"get_order_status": get_order_status}
38
39
    conversation = [{"role": "user", "content": [{"text": "What's the status of order ORD-123?"}]}]
40
41
    # Agent loop: call Bedrock, handle toolUse blocks, repeat
42
    MAX_STEPS = 5
43
    for _ in range(MAX_STEPS):
44
        converse_params = {
45
            "modelId": agent.model.name,
46
            "messages": conversation,
47
            "system": [{"text": agent.instructions}],
48
            "toolConfig": tool_config,
49
        }
50
        if inference:
51
            converse_params["inferenceConfig"] = inference
52
53
        response = tracker.track_bedrock_converse_metrics(
54
            bedrock_client.converse(**converse_params)
55
        )
56
57
        output_message = response.get("output", {}).get("message", {})
58
        stop_reason = response.get("stopReason")
59
60
        if stop_reason != "tool_use":
61
            break
62
63
        # Append assistant turn (preserves toolUse blocks)
64
        conversation.append(output_message)
65
66
        # Build a single tool-result user turn covering every toolUse block
67
        tool_results = []
68
        for block in output_message.get("content", []):
69
            if "toolUse" not in block:
70
                continue
71
            tool_use = block["toolUse"]
72
            if tool_use["name"] not in tool_handlers:
73
                raise ValueError(f"Unknown tool: {tool_use['name']}")
74
            result = tool_handlers[tool_use["name"]](**tool_use["input"])
75
            tracker.track_tool_call(tool_use["name"])
76
            tool_results.append({
77
                "toolResult": {
78
                    "toolUseId": tool_use["toolUseId"],
79
                    "content": [{"text": result}],
80
                }
81
            })
82
        conversation.append({"role": "user", "content": tool_results})

1
ldclient.get().flush()
2
ldclient.get().close()

1
ldclient.get().flush()
2
ldclient.get().close()

1
import os
2
import boto3
3
import ldclient
4
from ldclient import Context
5
from ldclient.config import Config
6
from ldai.client import LDAIClient, AICompletionConfigDefault, AIAgentConfigDefault
7
from dotenv import load_dotenv
8
9
load_dotenv()
10
11
SDK_KEY = os.environ.get("LAUNCHDARKLY_SDK_KEY_BEDROCK")
12
13
CONFIG_KEY = "bedrock-assistant"
14
AGENT_CONFIG_KEY = "bedrock-agent"
15
16
17
def inference_config_from_params(params):
18
    """Map AI Config parameters to Bedrock's Converse inferenceConfig shape.
19
    Rename max_tokens to maxTokens; keep temperature, topP, stopSequences as-is."""
20
    mapping = {"max_tokens": "maxTokens"}
21
    allowed = {"maxTokens", "temperature", "topP", "stopSequences"}
22
    renamed = {mapping.get(k, k): v for k, v in (params or {}).items()}
23
    return {k: v for k, v in renamed.items() if k in allowed}
24
25
26
def main():
27
    ldclient.set_config(Config(SDK_KEY))
28
    if not ldclient.get().is_initialized():
29
        return
30
31
    ai_client = LDAIClient(ldclient.get())
32
33
    bedrock_client = boto3.client(
34
        'bedrock-runtime',
35
        region_name=os.environ.get('AWS_REGION', 'us-west-2'),
36
    )
37
38
    context = Context.builder("user-123").kind("user").name("Sandy").build()
39
40
    # ===================
41
    # COMPLETION MODE
42
    # ===================
43
    # Pass a default for improved resiliency when the AI config is unavailable
44
    # or LaunchDarkly is unreachable; omit for a disabled default.
45
    # Example:
46
    #   default = AICompletionConfigDefault(
47
    #       enabled=True,
48
    #       model={"name": "us.anthropic.claude-sonnet-4-20250514-v1:0"},
49
    #       provider={"name": "bedrock"},
50
    #       messages=[{"role": "system", "content": "You are a helpful assistant."}],
51
    #   )
52
    #   config = ai_client.completion_config(CONFIG_KEY, context, default, variables={"topic": "Python"})
53
    config = ai_client.completion_config(CONFIG_KEY, context, variables={"topic": "Python"})
54
55
    if config.enabled:
56
        tracker = config.tracker
57
        messages = config.messages or []
58
59
        # Bedrock Converse: system messages are a top-level list, not a role
60
        system_messages = [{'text': m.content} for m in messages if m.role == 'system']
61
        conversation = [
62
            {'role': m.role, 'content': [{'text': m.content}]}
63
            for m in messages if m.role in ('user', 'assistant')
64
        ]
65
        conversation.append({
66
            'role': 'user',
67
            'content': [{'text': 'How do I read a file in Python?'}]
68
        })
69
70
        converse_params = {
71
            'modelId': config.model.name,
72
            'messages': conversation,
73
        }
74
        if system_messages:
75
            converse_params['system'] = system_messages
76
        ld_params = (config.model.to_dict().get("parameters") if config.model else None) or {}
77
        inference = inference_config_from_params(ld_params)
78
        if inference:
79
            converse_params['inferenceConfig'] = inference
80
81
        tracker.track_bedrock_converse_metrics(
82
            bedrock_client.converse(**converse_params)
83
        )
84
85
    # ===================
86
    # AGENT MODE
87
    # ===================
88
    agent = ai_client.agent_config(
89
        AGENT_CONFIG_KEY,
90
        context,
91
        AIAgentConfigDefault(enabled=False),
92
    )
93
94
    if agent.enabled:
95
        tracker = agent.tracker
96
        ld_params = (agent.model.to_dict().get("parameters") if agent.model else None) or {}
97
        inference = inference_config_from_params(ld_params)
98
99
        # LD returns attached tools under parameters.tools in OpenAI type=function shape.
100
        # Bedrock Converse wants {toolSpec: {name, description, inputSchema: {json}}}.
101
        ld_tools = ld_params.get("tools", []) or []
102
        tool_config = {
103
            "tools": [
104
                {
105
                    "toolSpec": {
106
                        "name": t["name"],
107
                        "description": t.get("description", ""),
108
                        "inputSchema": {"json": t.get("parameters", {"type": "object", "properties": {}})},
109
                    }
110
                }
111
                for t in ld_tools
112
            ]
113
        }
114
115
        # Handlers stay in application code — LD governs the schema, the app owns execution.
116
        def get_order_status(order_id: str) -> str:
117
            orders = {
118
                "ORD-123": "Shipped — arrives Thursday",
119
                "ORD-456": "Processing — estimated ship date: tomorrow",
120
                "ORD-789": "Delivered on Monday",
121
            }
122
            return orders.get(order_id, f"No order found with ID {order_id}")
123
124
        tool_handlers = {"get_order_status": get_order_status}
125
126
        conversation = [{"role": "user", "content": [{"text": "What's the status of order ORD-123?"}]}]
127
128
        # Agent loop: call Bedrock, handle toolUse blocks, repeat
129
        MAX_STEPS = 5
130
        for _ in range(MAX_STEPS):
131
            converse_params = {
132
                "modelId": agent.model.name,
133
                "messages": conversation,
134
                "system": [{"text": agent.instructions}],
135
                "toolConfig": tool_config,
136
            }
137
            if inference:
138
                converse_params["inferenceConfig"] = inference
139
140
            response = tracker.track_bedrock_converse_metrics(
141
                bedrock_client.converse(**converse_params)
142
            )
143
144
            output_message = response.get("output", {}).get("message", {})
145
            stop_reason = response.get("stopReason")
146
147
            if stop_reason != "tool_use":
148
                break
149
150
            # Append assistant turn (preserves toolUse blocks)
151
            conversation.append(output_message)
152
153
            # Build a single tool-result user turn covering every toolUse block
154
            tool_results = []
155
            for block in output_message.get("content", []):
156
                if "toolUse" not in block:
157
                    continue
158
                tool_use = block["toolUse"]
159
                if tool_use["name"] not in tool_handlers:
160
                    raise ValueError(f"Unknown tool: {tool_use['name']}")
161
                result = tool_handlers[tool_use["name"]](**tool_use["input"])
162
                tracker.track_tool_call(tool_use["name"])
163
                tool_results.append({
164
                    "toolResult": {
165
                        "toolUseId": tool_use["toolUseId"],
166
                        "content": [{"text": result}],
167
                    }
168
                })
169
            conversation.append({"role": "user", "content": tool_results})
170
171
    ldclient.get().flush()
172
    ldclient.get().close()
173
174
175
if __name__ == "__main__":
176
    main()