rboto-bedrock-agent-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 Agent Runtime, powered by Rust. This package contains BedrockAgentRuntimeClient and 35 operations.

Installation

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

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

You can also install rboto-bedrock-agent-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 BedrockAgentRuntimeClient
from rboto_bedrock_agent_runtime.types import KnowledgeBaseQuery


async def main() -> None:
    client = BedrockAgentRuntimeClient(region="us-east-1")
    query: KnowledgeBaseQuery = {"text": "How do I reset my password?"}
    response = await client.retrieve(knowledge_base_id="KB12345678", retrieval_query=query)
    for result in response.retrieval_results:
        print(result.content, result.score)


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 BedrockAgentRuntimeClient.

• 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_agent_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_agent_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_agent_runtime.types; native response definitions are in the package's _native.pyi.

Method	Required arguments	Return type
agentic_retrieve_stream	messages: Sequence[AgenticRetrieveMessage], retrievers: Sequence[AgenticRetriever], agentic_retrieve_configuration: AgenticRetrieveConfiguration	AgenticRetrieveStreamOutput
check_ingested_document_acl	knowledge_base_id: str, data_source_id: str, document_id: str, user_context: UserContext	CheckIngestedDocumentAclOutput
create_invocation	session_identifier: str	CreateInvocationOutput
create_session	None	CreateSessionOutput
delete_agent_memory	agent_id: str, agent_alias_id: str	DeleteAgentMemoryOutput
delete_session	session_identifier: str	DeleteSessionOutput
end_session	session_identifier: str	EndSessionOutput
generate_query	query_generation_input: QueryGenerationInput, transformation_configuration: TransformationConfiguration	GenerateQueryOutput
get_agent_memory	agent_id: str, agent_alias_id: str, memory_type: MemoryType, memory_id: str	GetAgentMemoryOutput
get_document_content	knowledge_base_id: str, data_source_id: str, document_id: str	GetDocumentContentOutput
get_execution_flow_snapshot	flow_identifier: str, flow_alias_identifier: str, execution_identifier: str	GetExecutionFlowSnapshotOutput
get_flow_execution	flow_identifier: str, flow_alias_identifier: str, execution_identifier: str	GetFlowExecutionOutput
get_ingested_document_acl	knowledge_base_id: str, data_source_id: str, document_id: str	GetIngestedDocumentAclOutput
get_invocation_step	invocation_identifier: str, invocation_step_id: str, session_identifier: str	GetInvocationStepOutput
get_session	session_identifier: str	GetSessionOutput
invoke_agent	agent_id: str, agent_alias_id: str, session_id: str	InvokeAgentOutput
invoke_flow	flow_identifier: str, flow_alias_identifier: str, inputs: Sequence[FlowInput]	InvokeFlowOutput
invoke_inline_agent	foundation_model: str, instruction: str, session_id: str	InvokeInlineAgentOutput
list_flow_execution_events	flow_identifier: str, flow_alias_identifier: str, execution_identifier: str, event_type: FlowExecutionEventType	ListFlowExecutionEventsOutput
list_flow_executions	flow_identifier: str	ListFlowExecutionsOutput
list_invocation_steps	session_identifier: str	ListInvocationStepsOutput
list_invocations	session_identifier: str	ListInvocationsOutput
list_sessions	None	ListSessionsOutput
list_tags_for_resource	resource_arn: str	ListTagsForResourceOutput
optimize_prompt	input: InputPrompt, target_model_id: str	OptimizePromptOutput
put_invocation_step	session_identifier: str, invocation_identifier: str, invocation_step_time: str, payload: InvocationStepPayload	PutInvocationStepOutput
rerank	queries: Sequence[RerankQuery], sources: Sequence[RerankSource], reranking_configuration: RerankingConfiguration	RerankOutput
retrieve	knowledge_base_id: str, retrieval_query: KnowledgeBaseQuery	RetrieveOutput
retrieve_and_generate	input: RetrieveAndGenerateInput	RetrieveAndGenerateOutput
retrieve_and_generate_stream	input: RetrieveAndGenerateInput	RetrieveAndGenerateStreamOutput
start_flow_execution	flow_identifier: str, flow_alias_identifier: str, inputs: Sequence[FlowInput]	StartFlowExecutionOutput
stop_flow_execution	flow_identifier: str, flow_alias_identifier: str, execution_identifier: str	StopFlowExecutionOutput
tag_resource	resource_arn: str, tags: dict[str, str]	TagResourceOutput
untag_resource	resource_arn: str, tag_keys: Sequence[str]	UntagResourceOutput
update_session	session_identifier: str	UpdateSessionOutput

Complete signatures

agentic_retrieve_stream

async def agentic_retrieve_stream(
    *,
    messages: Sequence[AgenticRetrieveMessage],
    retrievers: Sequence[AgenticRetriever],
    agentic_retrieve_configuration: AgenticRetrieveConfiguration,
    policy_configuration: AgenticRetrievePolicyConfiguration | None = None,
    next_token: str | None = None,
    user_context: UserContext | None = None,
    memory_configuration: AgenticRetrieveMemoryConfiguration | None = None,
    retrieval_instruction_configuration: AgenticRetrieveInstructionConfiguration | None = None,
    generate_response: bool | None = None,
) -> AgenticRetrieveStreamOutput: ...

check_ingested_document_acl

async def check_ingested_document_acl(
    *,
    knowledge_base_id: str,
    data_source_id: str,
    document_id: str,
    user_context: UserContext,
) -> CheckIngestedDocumentAclOutput: ...

create_invocation

async def create_invocation(
    *,
    session_identifier: str,
    invocation_id: str | None = None,
    description: str | None = None,
) -> CreateInvocationOutput: ...

create_session

async def create_session(
    *,
    session_metadata: dict[str, str] | None = None,
    encryption_key_arn: str | None = None,
    tags: dict[str, str] | None = None,
) -> CreateSessionOutput: ...

delete_agent_memory

async def delete_agent_memory(
    *,
    agent_id: str,
    agent_alias_id: str,
    memory_id: str | None = None,
    session_id: str | None = None,
) -> DeleteAgentMemoryOutput: ...

delete_session

async def delete_session(
    *,
    session_identifier: str,
) -> DeleteSessionOutput: ...

end_session

async def end_session(
    *,
    session_identifier: str,
) -> EndSessionOutput: ...

generate_query

async def generate_query(
    *,
    query_generation_input: QueryGenerationInput,
    transformation_configuration: TransformationConfiguration,
) -> GenerateQueryOutput: ...

get_agent_memory

async def get_agent_memory(
    *,
    agent_id: str,
    agent_alias_id: str,
    memory_type: MemoryType,
    memory_id: str,
    next_token: str | None = None,
    max_items: int | None = None,
) -> GetAgentMemoryOutput: ...

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

get_document_content

async def get_document_content(
    *,
    knowledge_base_id: str,
    data_source_id: str,
    document_id: str,
    output_format: DocumentOutputFormat | None = None,
    user_context: UserContext | None = None,
) -> GetDocumentContentOutput: ...

get_execution_flow_snapshot

async def get_execution_flow_snapshot(
    *,
    flow_identifier: str,
    flow_alias_identifier: str,
    execution_identifier: str,
) -> GetExecutionFlowSnapshotOutput: ...

get_flow_execution

async def get_flow_execution(
    *,
    flow_identifier: str,
    flow_alias_identifier: str,
    execution_identifier: str,
) -> GetFlowExecutionOutput: ...

get_ingested_document_acl

async def get_ingested_document_acl(
    *,
    knowledge_base_id: str,
    data_source_id: str,
    document_id: str,
) -> GetIngestedDocumentAclOutput: ...

get_invocation_step

async def get_invocation_step(
    *,
    invocation_identifier: str,
    invocation_step_id: str,
    session_identifier: str,
) -> GetInvocationStepOutput: ...

get_session

async def get_session(
    *,
    session_identifier: str,
) -> GetSessionOutput: ...

invoke_agent

async def invoke_agent(
    *,
    agent_id: str,
    agent_alias_id: str,
    session_id: str,
    session_state: SessionState | None = None,
    end_session: bool | None = None,
    enable_trace: bool | None = None,
    input_text: str | None = None,
    memory_id: str | None = None,
    bedrock_model_configurations: BedrockModelConfigurations | None = None,
    streaming_configurations: StreamingConfigurations | None = None,
    prompt_creation_configurations: PromptCreationConfigurations | None = None,
    source_arn: str | None = None,
) -> InvokeAgentOutput: ...

invoke_flow

async def invoke_flow(
    *,
    flow_identifier: str,
    flow_alias_identifier: str,
    inputs: Sequence[FlowInput],
    enable_trace: bool | None = None,
    model_performance_configuration: ModelPerformanceConfiguration | None = None,
    execution_id: str | None = None,
) -> InvokeFlowOutput: ...

invoke_inline_agent

async def invoke_inline_agent(
    *,
    foundation_model: str,
    instruction: str,
    session_id: str,
    customer_encryption_key_arn: str | None = None,
    idle_session_ttl_in_seconds: int | None = None,
    action_groups: Sequence[AgentActionGroup] | None = None,
    knowledge_bases: Sequence[KnowledgeBase] | None = None,
    guardrail_configuration: GuardrailConfigurationWithArn | None = None,
    prompt_override_configuration: PromptOverrideConfiguration | None = None,
    agent_collaboration: AgentCollaboration | None = None,
    collaborator_configurations: Sequence[CollaboratorConfiguration] | None = None,
    agent_name: str | None = None,
    end_session: bool | None = None,
    enable_trace: bool | None = None,
    input_text: str | None = None,
    streaming_configurations: StreamingConfigurations | None = None,
    prompt_creation_configurations: PromptCreationConfigurations | None = None,
    inline_session_state: InlineSessionState | None = None,
    collaborators: Sequence[Collaborator] | None = None,
    bedrock_model_configurations: InlineBedrockModelConfigurations | None = None,
    orchestration_type: OrchestrationType | None = None,
    custom_orchestration: CustomOrchestration | None = None,
) -> InvokeInlineAgentOutput: ...

list_flow_execution_events

async def list_flow_execution_events(
    *,
    flow_identifier: str,
    flow_alias_identifier: str,
    execution_identifier: str,
    event_type: FlowExecutionEventType,
    max_results: int | None = None,
    next_token: str | None = None,
) -> ListFlowExecutionEventsOutput: ...

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

list_flow_executions

async def list_flow_executions(
    *,
    flow_identifier: str,
    flow_alias_identifier: str | None = None,
    max_results: int | None = None,
    next_token: str | None = None,
) -> ListFlowExecutionsOutput: ...

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

list_invocation_steps

async def list_invocation_steps(
    *,
    session_identifier: str,
    invocation_identifier: str | None = None,
    next_token: str | None = None,
    max_results: int | None = None,
) -> ListInvocationStepsOutput: ...

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

list_invocations

async def list_invocations(
    *,
    session_identifier: str,
    next_token: str | None = None,
    max_results: int | None = None,
) -> ListInvocationsOutput: ...

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

list_sessions

async def list_sessions(
    *,
    max_results: int | None = None,
    next_token: str | None = None,
) -> ListSessionsOutput: ...

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

list_tags_for_resource

async def list_tags_for_resource(
    *,
    resource_arn: str,
) -> ListTagsForResourceOutput: ...

optimize_prompt

async def optimize_prompt(
    *,
    input: InputPrompt,
    target_model_id: str,
) -> OptimizePromptOutput: ...

put_invocation_step

async def put_invocation_step(
    *,
    session_identifier: str,
    invocation_identifier: str,
    invocation_step_time: str,
    payload: InvocationStepPayload,
    invocation_step_id: str | None = None,
) -> PutInvocationStepOutput: ...

rerank

async def rerank(
    *,
    queries: Sequence[RerankQuery],
    sources: Sequence[RerankSource],
    reranking_configuration: RerankingConfiguration,
    next_token: str | None = None,
) -> RerankOutput: ...

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

retrieve

async def retrieve(
    *,
    knowledge_base_id: str,
    retrieval_query: KnowledgeBaseQuery,
    retrieval_configuration: KnowledgeBaseRetrievalConfiguration | None = None,
    guardrail_configuration: GuardrailConfiguration | None = None,
    next_token: str | None = None,
    user_context: UserContext | None = None,
) -> RetrieveOutput: ...

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

retrieve_and_generate

async def retrieve_and_generate(
    *,
    input: RetrieveAndGenerateInput,
    session_id: str | None = None,
    retrieve_and_generate_configuration: RetrieveAndGenerateConfiguration | None = None,
    session_configuration: RetrieveAndGenerateSessionConfiguration | None = None,
    user_context: UserContext | None = None,
) -> RetrieveAndGenerateOutput: ...

retrieve_and_generate_stream

async def retrieve_and_generate_stream(
    *,
    input: RetrieveAndGenerateInput,
    session_id: str | None = None,
    retrieve_and_generate_configuration: RetrieveAndGenerateConfiguration | None = None,
    session_configuration: RetrieveAndGenerateSessionConfiguration | None = None,
    user_context: UserContext | None = None,
) -> RetrieveAndGenerateStreamOutput: ...

start_flow_execution

async def start_flow_execution(
    *,
    flow_identifier: str,
    flow_alias_identifier: str,
    inputs: Sequence[FlowInput],
    flow_execution_name: str | None = None,
    model_performance_configuration: ModelPerformanceConfiguration | None = None,
) -> StartFlowExecutionOutput: ...

stop_flow_execution

async def stop_flow_execution(
    *,
    flow_identifier: str,
    flow_alias_identifier: str,
    execution_identifier: str,
) -> StopFlowExecutionOutput: ...

tag_resource

async def tag_resource(
    *,
    resource_arn: str,
    tags: dict[str, str],
) -> TagResourceOutput: ...

untag_resource

async def untag_resource(
    *,
    resource_arn: str,
    tag_keys: Sequence[str],
) -> UntagResourceOutput: ...

update_session

async def update_session(
    *,
    session_identifier: str,
    session_metadata: dict[str, str] | None = None,
) -> UpdateSessionOutput: ...

Links

• Service guide
• Repository
• Issues
• Releases

Licensed under Apache-2.0.