rboto-kms

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 AWS KMS, powered by Rust. This package contains KMSClient and 54 operations.

Installation

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

python -m pip install "rboto[kms]"

You can also install rboto-kms 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 KMSClient


async def main() -> None:
    client = KMSClient(region="us-east-1")
    response = await client.encrypt(key_id="alias/application", plaintext=b"hello")
    print(response.ciphertext_blob)


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

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

Method	Required arguments	Return type
cancel_key_deletion	key_id: str	CancelKeyDeletionOutput
connect_custom_key_store	custom_key_store_id: str	ConnectCustomKeyStoreOutput
create_alias	alias_name: str, target_key_id: str	None
create_custom_key_store	custom_key_store_name: str	CreateCustomKeyStoreOutput
create_grant	key_id: str, operations: Sequence[GrantOperation]	CreateGrantOutput
create_key	None	CreateKeyOutput
decrypt	None	DecryptOutput
delete_alias	alias_name: str	None
delete_custom_key_store	custom_key_store_id: str	DeleteCustomKeyStoreOutput
delete_imported_key_material	key_id: str	DeleteImportedKeyMaterialOutput
derive_shared_secret	key_id: str, key_agreement_algorithm: KeyAgreementAlgorithmSpec, public_key: bytes	DeriveSharedSecretOutput
describe_custom_key_stores	None	DescribeCustomKeyStoresOutput
describe_key	key_id: str	DescribeKeyOutput
disable_key	key_id: str	None
disable_key_rotation	key_id: str	None
disconnect_custom_key_store	custom_key_store_id: str	DisconnectCustomKeyStoreOutput
enable_key	key_id: str	None
enable_key_rotation	key_id: str	None
encrypt	key_id: str, plaintext: bytes	EncryptOutput
generate_data_key	key_id: str	GenerateDataKeyOutput
generate_data_key_pair	key_id: str, key_pair_spec: DataKeyPairSpec	GenerateDataKeyPairOutput
generate_data_key_pair_without_plaintext	key_id: str, key_pair_spec: DataKeyPairSpec	GenerateDataKeyPairWithoutPlaintextOutput
generate_data_key_without_plaintext	key_id: str	GenerateDataKeyWithoutPlaintextOutput
generate_mac	message: bytes, key_id: str, mac_algorithm: MacAlgorithmSpec	GenerateMacOutput
generate_random	None	GenerateRandomOutput
get_key_last_usage	key_id: str	GetKeyLastUsageOutput
get_key_policy	key_id: str	GetKeyPolicyOutput
get_key_rotation_status	key_id: str	GetKeyRotationStatusOutput
get_parameters_for_import	key_id: str, wrapping_algorithm: AlgorithmSpec, wrapping_key_spec: WrappingKeySpec	GetParametersForImportOutput
get_public_key	key_id: str	GetPublicKeyOutput
import_key_material	key_id: str, import_token: bytes, encrypted_key_material: bytes	ImportKeyMaterialOutput
list_aliases	None	ListAliasesOutput
list_grants	key_id: str	ListGrantsOutput
list_key_policies	key_id: str	ListKeyPoliciesOutput
list_key_rotations	key_id: str	ListKeyRotationsOutput
list_keys	None	ListKeysOutput
list_resource_tags	key_id: str	ListResourceTagsOutput
list_retirable_grants	None	ListRetirableGrantsOutput
put_key_policy	key_id: str, policy: str	None
re_encrypt	destination_key_id: str	ReEncryptOutput
replicate_key	key_id: str, replica_region: str	ReplicateKeyOutput
retire_grant	None	None
revoke_grant	key_id: str, grant_id: str	None
rotate_key_on_demand	key_id: str	RotateKeyOnDemandOutput
schedule_key_deletion	key_id: str	ScheduleKeyDeletionOutput
sign	key_id: str, message: bytes, signing_algorithm: SigningAlgorithmSpec	SignOutput
tag_resource	key_id: str, tags: Sequence[Tag]	None
untag_resource	key_id: str, tag_keys: Sequence[str]	None
update_alias	alias_name: str, target_key_id: str	None
update_custom_key_store	custom_key_store_id: str	UpdateCustomKeyStoreOutput
update_key_description	key_id: str, description: str	None
update_primary_region	key_id: str, primary_region: str	None
verify	key_id: str, message: bytes, signature: bytes, signing_algorithm: SigningAlgorithmSpec	VerifyOutput
verify_mac	message: bytes, key_id: str, mac_algorithm: MacAlgorithmSpec, mac: bytes	VerifyMacOutput

Complete signatures

cancel_key_deletion

async def cancel_key_deletion(
    *,
    key_id: str,
) -> CancelKeyDeletionOutput: ...

connect_custom_key_store

async def connect_custom_key_store(
    *,
    custom_key_store_id: str,
) -> ConnectCustomKeyStoreOutput: ...

create_alias

async def create_alias(
    *,
    alias_name: str,
    target_key_id: str,
) -> None: ...

create_custom_key_store

async def create_custom_key_store(
    *,
    custom_key_store_name: str,
    cloud_hsm_cluster_id: str | None = None,
    trust_anchor_certificate: str | None = None,
    key_store_password: str | None = None,
    custom_key_store_type: CustomKeyStoreType | None = None,
    xks_proxy_uri_endpoint: str | None = None,
    xks_proxy_uri_path: str | None = None,
    xks_proxy_vpc_endpoint_service_name: str | None = None,
    xks_proxy_vpc_endpoint_service_owner: str | None = None,
    xks_proxy_authentication_credential: XksProxyAuthenticationCredentialType | None = None,
    xks_proxy_connectivity: XksProxyConnectivityType | None = None,
) -> CreateCustomKeyStoreOutput: ...

create_grant

async def create_grant(
    *,
    key_id: str,
    operations: Sequence[GrantOperation],
    grantee_principal: str | None = None,
    retiring_principal: str | None = None,
    constraints: GrantConstraints | None = None,
    grant_tokens: Sequence[str] | None = None,
    name: str | None = None,
    dry_run: bool | None = None,
    grantee_service_principal: str | None = None,
    retiring_service_principal: str | None = None,
) -> CreateGrantOutput: ...

create_key

async def create_key(
    *,
    policy: str | None = None,
    description: str | None = None,
    key_usage: KeyUsageType | None = None,
    customer_master_key_spec: CustomerMasterKeySpec | None = None,
    key_spec: KeySpec | None = None,
    origin: OriginType | None = None,
    custom_key_store_id: str | None = None,
    bypass_policy_lockout_safety_check: bool | None = None,
    tags: Sequence[Tag] | None = None,
    multi_region: bool | None = None,
    xks_key_id: str | None = None,
) -> CreateKeyOutput: ...

decrypt

async def decrypt(
    *,
    ciphertext_blob: bytes | None = None,
    encryption_context: dict[str, str] | None = None,
    grant_tokens: Sequence[str] | None = None,
    key_id: str | None = None,
    encryption_algorithm: EncryptionAlgorithmSpec | None = None,
    recipient: RecipientInfo | None = None,
    dry_run: bool | None = None,
    dry_run_modifiers: Sequence[DryRunModifierType] | None = None,
) -> DecryptOutput: ...

delete_alias

async def delete_alias(
    *,
    alias_name: str,
) -> None: ...

delete_custom_key_store

async def delete_custom_key_store(
    *,
    custom_key_store_id: str,
) -> DeleteCustomKeyStoreOutput: ...

delete_imported_key_material

async def delete_imported_key_material(
    *,
    key_id: str,
    key_material_id: str | None = None,
) -> DeleteImportedKeyMaterialOutput: ...

derive_shared_secret

async def derive_shared_secret(
    *,
    key_id: str,
    key_agreement_algorithm: KeyAgreementAlgorithmSpec,
    public_key: bytes,
    grant_tokens: Sequence[str] | None = None,
    dry_run: bool | None = None,
    recipient: RecipientInfo | None = None,
) -> DeriveSharedSecretOutput: ...

describe_custom_key_stores

async def describe_custom_key_stores(
    *,
    custom_key_store_id: str | None = None,
    custom_key_store_name: str | None = None,
    limit: int | None = None,
    marker: str | None = None,
) -> DescribeCustomKeyStoresOutput: ...

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

describe_key

async def describe_key(
    *,
    key_id: str,
    grant_tokens: Sequence[str] | None = None,
) -> DescribeKeyOutput: ...

disable_key

async def disable_key(
    *,
    key_id: str,
) -> None: ...

disable_key_rotation

async def disable_key_rotation(
    *,
    key_id: str,
) -> None: ...

disconnect_custom_key_store

async def disconnect_custom_key_store(
    *,
    custom_key_store_id: str,
) -> DisconnectCustomKeyStoreOutput: ...

enable_key

async def enable_key(
    *,
    key_id: str,
) -> None: ...

enable_key_rotation

async def enable_key_rotation(
    *,
    key_id: str,
    rotation_period_in_days: int | None = None,
) -> None: ...

encrypt

async def encrypt(
    *,
    key_id: str,
    plaintext: bytes,
    encryption_context: dict[str, str] | None = None,
    grant_tokens: Sequence[str] | None = None,
    encryption_algorithm: EncryptionAlgorithmSpec | None = None,
    dry_run: bool | None = None,
) -> EncryptOutput: ...

generate_data_key

async def generate_data_key(
    *,
    key_id: str,
    encryption_context: dict[str, str] | None = None,
    number_of_bytes: int | None = None,
    key_spec: DataKeySpec | None = None,
    grant_tokens: Sequence[str] | None = None,
    recipient: RecipientInfo | None = None,
    dry_run: bool | None = None,
) -> GenerateDataKeyOutput: ...

generate_data_key_pair

async def generate_data_key_pair(
    *,
    key_id: str,
    key_pair_spec: DataKeyPairSpec,
    encryption_context: dict[str, str] | None = None,
    grant_tokens: Sequence[str] | None = None,
    recipient: RecipientInfo | None = None,
    dry_run: bool | None = None,
) -> GenerateDataKeyPairOutput: ...

generate_data_key_pair_without_plaintext

async def generate_data_key_pair_without_plaintext(
    *,
    key_id: str,
    key_pair_spec: DataKeyPairSpec,
    encryption_context: dict[str, str] | None = None,
    grant_tokens: Sequence[str] | None = None,
    dry_run: bool | None = None,
) -> GenerateDataKeyPairWithoutPlaintextOutput: ...

generate_data_key_without_plaintext

async def generate_data_key_without_plaintext(
    *,
    key_id: str,
    encryption_context: dict[str, str] | None = None,
    key_spec: DataKeySpec | None = None,
    number_of_bytes: int | None = None,
    grant_tokens: Sequence[str] | None = None,
    dry_run: bool | None = None,
) -> GenerateDataKeyWithoutPlaintextOutput: ...

generate_mac

async def generate_mac(
    *,
    message: bytes,
    key_id: str,
    mac_algorithm: MacAlgorithmSpec,
    grant_tokens: Sequence[str] | None = None,
    dry_run: bool | None = None,
) -> GenerateMacOutput: ...

generate_random

async def generate_random(
    *,
    number_of_bytes: int | None = None,
    custom_key_store_id: str | None = None,
    recipient: RecipientInfo | None = None,
) -> GenerateRandomOutput: ...

get_key_last_usage

async def get_key_last_usage(
    *,
    key_id: str,
) -> GetKeyLastUsageOutput: ...

get_key_policy

async def get_key_policy(
    *,
    key_id: str,
    policy_name: str | None = None,
) -> GetKeyPolicyOutput: ...

get_key_rotation_status

async def get_key_rotation_status(
    *,
    key_id: str,
) -> GetKeyRotationStatusOutput: ...

get_parameters_for_import

async def get_parameters_for_import(
    *,
    key_id: str,
    wrapping_algorithm: AlgorithmSpec,
    wrapping_key_spec: WrappingKeySpec,
) -> GetParametersForImportOutput: ...

get_public_key

async def get_public_key(
    *,
    key_id: str,
    grant_tokens: Sequence[str] | None = None,
) -> GetPublicKeyOutput: ...

import_key_material

async def import_key_material(
    *,
    key_id: str,
    import_token: bytes,
    encrypted_key_material: bytes,
    valid_to: str | None = None,
    expiration_model: ExpirationModelType | None = None,
    import_type: ImportType | None = None,
    key_material_description: str | None = None,
    key_material_id: str | None = None,
) -> ImportKeyMaterialOutput: ...

list_aliases

async def list_aliases(
    *,
    key_id: str | None = None,
    limit: int | None = None,
    marker: str | None = None,
) -> ListAliasesOutput: ...

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

list_grants

async def list_grants(
    *,
    key_id: str,
    limit: int | None = None,
    marker: str | None = None,
    grant_id: str | None = None,
    grantee_principal: str | None = None,
    grantee_service_principal: str | None = None,
) -> ListGrantsOutput: ...

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

list_key_policies

async def list_key_policies(
    *,
    key_id: str,
    limit: int | None = None,
    marker: str | None = None,
) -> ListKeyPoliciesOutput: ...

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

list_key_rotations

async def list_key_rotations(
    *,
    key_id: str,
    include_key_material: IncludeKeyMaterial | None = None,
    limit: int | None = None,
    marker: str | None = None,
) -> ListKeyRotationsOutput: ...

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

list_keys

async def list_keys(
    *,
    limit: int | None = None,
    marker: str | None = None,
) -> ListKeysOutput: ...

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

list_resource_tags

async def list_resource_tags(
    *,
    key_id: str,
    limit: int | None = None,
    marker: str | None = None,
) -> ListResourceTagsOutput: ...

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

list_retirable_grants

async def list_retirable_grants(
    *,
    limit: int | None = None,
    marker: str | None = None,
    retiring_principal: str | None = None,
    retiring_service_principal: str | None = None,
) -> ListRetirableGrantsOutput: ...

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

put_key_policy

async def put_key_policy(
    *,
    key_id: str,
    policy: str,
    policy_name: str | None = None,
    bypass_policy_lockout_safety_check: bool | None = None,
) -> None: ...

re_encrypt

async def re_encrypt(
    *,
    destination_key_id: str,
    ciphertext_blob: bytes | None = None,
    source_encryption_context: dict[str, str] | None = None,
    source_key_id: str | None = None,
    destination_encryption_context: dict[str, str] | None = None,
    source_encryption_algorithm: EncryptionAlgorithmSpec | None = None,
    destination_encryption_algorithm: EncryptionAlgorithmSpec | None = None,
    grant_tokens: Sequence[str] | None = None,
    dry_run: bool | None = None,
    dry_run_modifiers: Sequence[DryRunModifierType] | None = None,
) -> ReEncryptOutput: ...

replicate_key

async def replicate_key(
    *,
    key_id: str,
    replica_region: str,
    policy: str | None = None,
    bypass_policy_lockout_safety_check: bool | None = None,
    description: str | None = None,
    tags: Sequence[Tag] | None = None,
) -> ReplicateKeyOutput: ...

retire_grant

async def retire_grant(
    *,
    grant_token: str | None = None,
    key_id: str | None = None,
    grant_id: str | None = None,
    dry_run: bool | None = None,
) -> None: ...

revoke_grant

async def revoke_grant(
    *,
    key_id: str,
    grant_id: str,
    dry_run: bool | None = None,
) -> None: ...

rotate_key_on_demand

async def rotate_key_on_demand(
    *,
    key_id: str,
) -> RotateKeyOnDemandOutput: ...

schedule_key_deletion

async def schedule_key_deletion(
    *,
    key_id: str,
    pending_window_in_days: int | None = None,
) -> ScheduleKeyDeletionOutput: ...

sign

async def sign(
    *,
    key_id: str,
    message: bytes,
    signing_algorithm: SigningAlgorithmSpec,
    message_type: MessageType | None = None,
    grant_tokens: Sequence[str] | None = None,
    dry_run: bool | None = None,
) -> SignOutput: ...

tag_resource

async def tag_resource(
    *,
    key_id: str,
    tags: Sequence[Tag],
) -> None: ...

untag_resource

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

update_alias

async def update_alias(
    *,
    alias_name: str,
    target_key_id: str,
) -> None: ...

update_custom_key_store

async def update_custom_key_store(
    *,
    custom_key_store_id: str,
    new_custom_key_store_name: str | None = None,
    key_store_password: str | None = None,
    cloud_hsm_cluster_id: str | None = None,
    xks_proxy_uri_endpoint: str | None = None,
    xks_proxy_uri_path: str | None = None,
    xks_proxy_vpc_endpoint_service_name: str | None = None,
    xks_proxy_vpc_endpoint_service_owner: str | None = None,
    xks_proxy_authentication_credential: XksProxyAuthenticationCredentialType | None = None,
    xks_proxy_connectivity: XksProxyConnectivityType | None = None,
) -> UpdateCustomKeyStoreOutput: ...

update_key_description

async def update_key_description(
    *,
    key_id: str,
    description: str,
) -> None: ...

update_primary_region

async def update_primary_region(
    *,
    key_id: str,
    primary_region: str,
) -> None: ...

verify

async def verify(
    *,
    key_id: str,
    message: bytes,
    signature: bytes,
    signing_algorithm: SigningAlgorithmSpec,
    message_type: MessageType | None = None,
    grant_tokens: Sequence[str] | None = None,
    dry_run: bool | None = None,
) -> VerifyOutput: ...

verify_mac

async def verify_mac(
    *,
    message: bytes,
    key_id: str,
    mac_algorithm: MacAlgorithmSpec,
    mac: bytes,
    grant_tokens: Sequence[str] | None = None,
    dry_run: bool | None = None,
) -> VerifyMacOutput: ...

Links

• Service guide
• Repository
• Issues
• Releases

Licensed under Apache-2.0.