Amazon Bedrock

Amazon Bedrock models can be called through the Bedrock Runtime Converse and Converse Stream APIs using the actions below.

Install the optional extra (pulls boto3):

pip install "apache-burr[bedrock]"

Available Actions

BedrockAction is a single-step action for standard request/response flows. BedrockStreamingAction is a streaming action that emits chunks during execution and applies final state updates when complete.

Both actions accept:

  • model_id: Bedrock model identifier (for example Claude models on Bedrock)

  • input_mapper: maps Burr state to Bedrock messages (and optional system)

  • reads / writes: standard Burr state wiring

  • Optional AWS settings: region, max_retries, optional injected client

  • Optional guardrails: guardrail_id + required guardrail_version

  • Optional inference_config passed through to Bedrock

Note

If guardrail_id is set, guardrail_version is required.

Supported Models

Burr passes model_id directly to Bedrock Runtime. Any model that supports the Bedrock Converse / ConverseStream API in your account and region can be used. See AWS model availability by region and account entitlement:

IAM Permissions

At minimum, the runtime identity used by Burr should have permission to invoke the specific Bedrock model(s) you choose:

  • bedrock:InvokeModel

  • bedrock:InvokeModelWithResponseStream (for streaming)

Depending on your setup, you may also need permissions for guardrails and cross-account resource access. Restrict resources to specific model ARNs whenever possible.

Quick Example

from burr.integrations.bedrock import BedrockAction

def map_state_to_prompt(state):
    return {
        "messages": [{"role": "user", "content": [{"text": state["question"]}]}],
        "system": [{"text": "You are a concise assistant."}],
    }

ask_model = BedrockAction(
    name="ask_model",
    model_id="anthropic.claude-3-haiku-20240307-v1:0",
    input_mapper=map_state_to_prompt,
    reads=["question"],
    writes=["response", "usage", "stop_reason"],
    region="us-east-1",
)

See the runnable integration example in the repository: examples/integrations/bedrock.

class burr.integrations.bedrock.BedrockAction(
model_id: str,
input_mapper: StateToPromptMapper,
reads: list[str],
writes: list[str],
name: str = 'bedrock_invoke',
region: str | None = None,
guardrail_id: str | None = None,
guardrail_version: str | None = None,
inference_config: dict[str, Any] | None = None,
max_retries: int = 3,
client: Any | None = None,
)

Bases: _BedrockBase, SingleStepAction

Action that invokes Amazon Bedrock models using the Converse API.

Parameters:

  • model_id – Bedrock model identifier (e.g. Anthropic Claude on Bedrock).

  • input_mapper – Callable mapping State to Bedrock messages / optional system keys.

  • reads – State keys this action reads.

  • writes – State keys to update (typically include response).

  • name – Action name for the graph.

  • region – AWS region for the Bedrock runtime client (optional).

  • guardrail_id – If set, guardrail_version must also be set explicitly.

  • guardrail_version – Guardrail version string (use DRAFT only when intended).

  • inference_config – Passed as inferenceConfig; if omitted, defaults to a maxTokens limit. Pass an empty dict explicitly to send an empty config.

  • max_retries – Botocore retry configuration for the runtime client.

  • client – Optional pre-built bedrock-runtime client (for tests or injection).

Use run_and_update() to run the model and merge outputs into state.

run_and_update(
state: State,
**run_kwargs,
) → tuple[dict, State]

Performs a run/update at the same time.

Parameters:

  • state – State to run the action on

  • run_kwargs – Additional arguments to the function passed at runtime.

Returns:

Result of the action and the new state

class burr.integrations.bedrock.BedrockStreamingAction(
model_id: str,
input_mapper: StateToPromptMapper,
reads: list[str],
writes: list[str],
name: str = 'bedrock_stream',
region: str | None = None,
guardrail_id: str | None = None,
guardrail_version: str | None = None,
inference_config: dict[str, Any] | None = None,
max_retries: int = 3,
client: Any | None = None,
)

Bases: _BedrockBase, StreamingAction

Streaming Bedrock action using the Converse Stream API.

Parameters match BedrockAction except the default name is bedrock_stream. Yields chunk dicts from stream_run() and merges the final response in update().

stream_run(
state: State,
**run_kwargs,
) → Generator[dict, None, None]

Streaming action stream_run is different than standard action run. It: 1. streams in an intermediate result (the dict output) 2. yields the final result at the end

Note that the user, in this case, is responsible for joining the result.

For instance, you could have:

def stream_run(state: State) -> Generator[dict, None, dict]:
    buffer = [] # you might want to be more efficient than simple strcat
    for token in query(state['prompt']):
        yield {'response' : token}
        buffer.append(token)
    yield {'response' : "".join(buffer)}

This would utilize a simple string buffer (implemented by a list) to store the results and then join them at the end. We return the final result.

Parameters:

  • state – State to run the action on

  • run_kwargs – parameters passed to the run function – these are specified by inputs

Returns:

A generator that streams in a result and returns the final result

update(result: dict, state: State) → State

Performs a state update given a result and the current state. Returns a new, modified State (recall state is immutable – simply changing the state in place will not work).

In the context of Burr, this is only applied in the two-step actions, where the run and update() functions are separate. The function-based APIs for Burr use the SingleStepAction class, which performs them both at once. This is not (yet) exposed as an interface for users to extend.

Parameters:

  • result – Result of a function executing on the state

  • state – State to update

Returns:

A new, modified state.