rboto-bedrock-runtime

rboto is an independent project maintained by Ferrumio. It is not affiliated with, sponsored by, or endorsed by Amazon Web Services (AWS).

Preview (Alpha). rboto is under active development. APIs may change between releases.

rboto provides an async-only, typed Python client for Amazon Bedrock Runtime, powered by Rust. This package contains BedrockRuntimeClient and 11 operations.

Installation

Requires Python 3.12 or newer and a native wheel for your Python version and platform.

python -m pip install "rboto[bedrock-runtime]"

You can also install rboto-bedrock-runtime directly; it installs the matching rboto package. Only the selected service is installed.

These are the package names to use when a release is available on your package index. If your index has no release yet, follow the source installation instructions.

Usage

Configure credentials through your environment, shared profile, or workload role. The example uses us-east-1; select a region where your service and resources are available. Resource names, identifiers, input files, and model IDs in the example are placeholders to replace with your own.

import asyncio

from rboto import BedrockRuntimeClient
from rboto_bedrock_runtime.types import Message


async def main() -> None:
    client = BedrockRuntimeClient(region="us-east-1")
    message: Message = {"role": "user", "content": [{"text": "Explain async I/O briefly."}]}
    response = await client.converse(model_id="YOUR_MODEL_ID", messages=[message])
    print(response.output)
    print(response.usage)


if __name__ == "__main__":
    asyncio.run(main())

In an application with an event loop, call await main() instead of asyncio.run(main()). Reuse the client across calls. Constructing a client does not send a request.

API conventions

• Import the client with from rboto import BedrockRuntimeClient.

• Every operation is asynchronous: use await client.method(...).

• Operation arguments are keyword-only and use the Python names listed below.

• Structured inputs use typed dictionaries and tagged unions from rboto_bedrock_runtime.types.

• Regular responses expose typed attributes; response.to_dict() converts them to a dictionary.

• Event-stream responses may be typed dictionaries containing async iterators. Follow the declared return type; consume event streams with async for.

• Byte streams are consumed asynchronously. await stream.read() buffers the complete body; iterate the stream when processing a large response.

• Direct operation calls return one page.

• Use async for page in client.paginators.method(...) for supported paginated operations. Paginators accept the same typed arguments as the corresponding client method.

• Catch modeled exceptions from rboto_bedrock_runtime.exceptions. Their error_code, message, and operation_name identify service failures.

Clients also accept profile, endpoint_url, or shared Session settings. See configuration, errors, and streaming.

Operations

This is the complete operation catalog for this generated package. “Required arguments” lists arguments without defaults in the Python signature. Service-side conditions may require additional arguments. Each method links to its complete signature, including optional arguments. Nested input definitions are in rboto_bedrock_runtime.types; native response definitions are in the package's _native.pyi.

Complete signatures

apply_guardrail

async def apply_guardrail(
    *,
    guardrail_identifier: str,
    guardrail_version: str,
    source: GuardrailContentSource,
    content: Sequence[GuardrailContentBlock],
    output_scope: GuardrailOutputScope | None = None,
) -> ApplyGuardrailOutput: ...

converse

async def converse(
    *,
    model_id: str,
    messages: Sequence[Message] | None = None,
    system: Sequence[SystemContentBlock] | None = None,
    inference_config: InferenceConfiguration | None = None,
    tool_config: ToolConfiguration | None = None,
    guardrail_config: GuardrailConfiguration | None = None,
    additional_model_request_fields: JSONValue | None = None,
    prompt_variables: dict[str, PromptVariableValues] | None = None,
    additional_model_response_field_paths: Sequence[str] | None = None,
    request_metadata: dict[str, str] | None = None,
    performance_config: PerformanceConfiguration | None = None,
    service_tier: ServiceTier | None = None,
    output_config: OutputConfig | None = None,
) -> ConverseOutput: ...

converse_stream

async def converse_stream(
    *,
    model_id: str,
    messages: Sequence[Message] | None = None,
    system: Sequence[SystemContentBlock] | None = None,
    inference_config: InferenceConfiguration | None = None,
    tool_config: ToolConfiguration | None = None,
    guardrail_config: GuardrailStreamConfiguration | None = None,
    additional_model_request_fields: JSONValue | None = None,
    prompt_variables: dict[str, PromptVariableValues] | None = None,
    additional_model_response_field_paths: Sequence[str] | None = None,
    request_metadata: dict[str, str] | None = None,
    performance_config: PerformanceConfiguration | None = None,
    service_tier: ServiceTier | None = None,
    output_config: OutputConfig | None = None,
) -> ConverseStreamOutput: ...

count_tokens

async def count_tokens(
    *,
    model_id: str,
    input: CountTokensInput,
) -> CountTokensOutput: ...

get_async_invoke

async def get_async_invoke(
    *,
    invocation_arn: str,
) -> GetAsyncInvokeOutput: ...

invoke_guardrail_checks

async def invoke_guardrail_checks(
    *,
    messages: Sequence[GuardrailChecksMessage],
    checks: GuardrailChecksConfig,
) -> InvokeGuardrailChecksOutput: ...

invoke_model

async def invoke_model(
    *,
    model_id: str,
    body: bytes | None = None,
    content_type: str | None = None,
    accept: str | None = None,
    trace: Trace | None = None,
    guardrail_identifier: str | None = None,
    guardrail_version: str | None = None,
    performance_config_latency: PerformanceConfigLatency | None = None,
    service_tier: ServiceTierType | None = None,
    request_metadata: str | None = None,
) -> InvokeModelOutput: ...

invoke_model_with_bidirectional_stream

async def invoke_model_with_bidirectional_stream(
    *,
    model_id: str,
    body: AsyncIterator[InvokeModelWithBidirectionalStreamInput],
) -> InvokeModelWithBidirectionalStreamOutput: ...

invoke_model_with_response_stream

async def invoke_model_with_response_stream(
    *,
    model_id: str,
    body: bytes | None = None,
    content_type: str | None = None,
    accept: str | None = None,
    trace: Trace | None = None,
    guardrail_identifier: str | None = None,
    guardrail_version: str | None = None,
    performance_config_latency: PerformanceConfigLatency | None = None,
    service_tier: ServiceTierType | None = None,
    request_metadata: str | None = None,
) -> InvokeModelWithResponseStreamOutput: ...

list_async_invokes

async def list_async_invokes(
    *,
    submit_time_after: str | None = None,
    submit_time_before: str | None = None,
    status_equals: AsyncInvokeStatus | None = None,
    max_results: int | None = None,
    next_token: str | None = None,
    sort_by: SortAsyncInvocationBy | None = None,
    sort_order: SortOrder | None = None,
) -> ListAsyncInvokesOutput: ...

Also available as client.paginators.list_async_invokes(...), returning an async iterator of the response type above.

start_async_invoke

async def start_async_invoke(
    *,
    model_id: str,
    model_input: JSONValue,
    output_data_config: AsyncInvokeOutputDataConfig,
    client_request_token: str | None = None,
    tags: Sequence[Tag] | None = None,
) -> StartAsyncInvokeOutput: ...

Links

• Service guide
• Repository
• Issues
• Releases

Licensed under Apache-2.0.