Application services

Install the relevant client packages from the checkout; see Getting started. Examples use existing resources and require permission for the requested operation.

Service	Client	Examples
CloudWatch Logs	`CloudWatchLogsClient()`	Log groups, streams, queries, Live Tail
Systems Manager	`SSMClient()`	Parameter Store, documents, commands, automation
Key Management Service	`KMSClient()`	Encryption, decryption, signatures, keys
Step Functions	`StepFunctionsClient()`	State machines and workflow executions
Simple Email Service v2	`SESV2Client()`	Email content, templates, identities, configuration

CloudWatch Logs

Send events to an existing log group and stream:

import time

from rboto import CloudWatchLogsClient
from rboto_cloudwatch_logs.types import InputLogEvent


async def write_log(group: str, stream: str, message: str) -> None:
    event: InputLogEvent = {"timestamp": int(time.time() * 1000), "message": message}
    response = await CloudWatchLogsClient(region="us-east-1").put_log_events(
        log_group_name=group,
        log_stream_name=stream,
        log_events=[event],
    )
    if response.rejected_log_events_info is not None:
        raise RuntimeError("One or more log events were rejected")

Live Tail exposes a typed asynchronous event stream. Stream-specific failures raise SessionStreamingError or SessionTimeoutError from rboto_cloudwatch_logs.exceptions.

from collections.abc import AsyncIterator

from rboto import CloudWatchLogsClient
from rboto_cloudwatch_logs.types import StartLiveTailResponseStream


async def tail_logs(group_arn: str) -> AsyncIterator[StartLiveTailResponseStream]:
    response = await CloudWatchLogsClient(region="us-east-1").start_live_tail(
        log_group_identifiers=[group_arn],
    )
    async for event in response["response_stream"]:
        yield event

Systems Manager Parameter Store

from rboto import SSMClient


async def read_parameter(name: str) -> str | None:
    response = await SSMClient(region="us-east-1").get_parameter(
        name=name,
        with_decryption=True,
    )
    return response.parameter.value if response.parameter is not None else None

Missing parameters raise rboto_ssm.exceptions.ParameterNotFoundError. A parameter version or label can be appended to the name, such as /app/config:3.

KMS

Binary inputs and outputs use bytes; rboto handles their wire encoding.

from rboto import KMSClient


async def encrypt_value(key_id: str, value: bytes) -> bytes:
    response = await KMSClient(region="us-east-1").encrypt(
        key_id=key_id,
        plaintext=value,
        encryption_context={"application": "example"},
    )
    if response.ciphertext_blob is None:
        raise RuntimeError("KMS returned no ciphertext")
    return response.ciphertext_blob

Supply the same encryption context when decrypting the value.

Step Functions

Step Functions uses the sfn service name. Workflow inputs and outputs are JSON strings.

import json

from rboto import StepFunctionsClient


async def start_workflow(state_machine_arn: str, order_id: str) -> str:
    response = await StepFunctionsClient(region="us-east-1").start_execution(
        state_machine_arn=state_machine_arn,
        input=json.dumps({"order_id": order_id}),
    )
    return response.execution_arn

For synchronous Express executions, use start_sync_execution() and check status. A failed workflow can return a normal API response with status == "FAILED" and error/cause populated.

SES v2

from rboto import SESV2Client
from rboto_sesv2.types import EmailContent


async def send_message(sender: str, recipient: str, text: str) -> str | None:
    content: EmailContent = {
        "simple": {
            "subject": {"data": "Notification", "charset": "UTF-8"},
            "body": {"text": {"data": text, "charset": "UTF-8"}},
        }
    }
    response = await SESV2Client(region="us-east-1").send_email(
        from_email_address=sender,
        destination={"to_addresses": [recipient]},
        content=content,
    )
    return response.message_id

The sender must be an identity allowed by SES. Account sandbox and sending restrictions still apply. Attachments and raw messages accept binary content as bytes.