rboto-connecthealth

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 Connect Health, powered by Rust. This package contains ConnectHealthClient and 16 operations.

Installation

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

python -m pip install "rboto[connecthealth]"

You can also install rboto-connecthealth 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 ConnectHealthClient


async def main() -> None:
    client = ConnectHealthClient(region="us-east-1")
    response = await client.list_domains(max_results=10)
    for domain in response.domains:
        print(domain.domain_id)


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

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

Method	Required arguments	Return type
activate_subscription	domain_id: str, subscription_id: str	ActivateSubscriptionOutput
create_domain	name: str	CreateDomainOutput
create_subscription	domain_id: str	CreateSubscriptionOutput
deactivate_subscription	domain_id: str, subscription_id: str	DeactivateSubscriptionOutput
delete_domain	domain_id: str	DeleteDomainOutput
get_domain	domain_id: str	GetDomainOutput
get_medical_scribe_listening_session	session_id: str, domain_id: str, subscription_id: str	GetMedicalScribeListeningSessionOutput
get_patient_insights_job	domain_id: str, job_id: str	GetPatientInsightsJobOutput
get_subscription	domain_id: str, subscription_id: str	GetSubscriptionOutput
list_domains	None	ListDomainsOutput
list_subscriptions	domain_id: str	ListSubscriptionsOutput
list_tags_for_resource	resource_arn: str	ListTagsForResourceOutput
start_medical_scribe_listening_session	session_id: str, domain_id: str, subscription_id: str, language_code: MedicalScribeLanguageCode, media_sample_rate_hertz: int, media_encoding: MedicalScribeMediaEncoding	StartMedicalScribeListeningSessionOutput
start_patient_insights_job	domain_id: str, patient_context: PatientInsightsPatientContext, insights_context: InsightsContext, encounter_context: PatientInsightsEncounterContext, user_context: UserContext, input_data_config: InputDataConfig, output_data_config: OutputDataConfig	StartPatientInsightsJobOutput
tag_resource	resource_arn: str, tags: dict[str, str]	None
untag_resource	resource_arn: str, tag_keys: Sequence[str]	None

Complete signatures

activate_subscription

async def activate_subscription(
    *,
    domain_id: str,
    subscription_id: str,
) -> ActivateSubscriptionOutput: ...

create_domain

async def create_domain(
    *,
    name: str,
    kms_key_arn: str | None = None,
    web_app_setup_configuration: CreateWebAppConfiguration | None = None,
    tags: dict[str, str] | None = None,
) -> CreateDomainOutput: ...

create_subscription

async def create_subscription(
    *,
    domain_id: str,
) -> CreateSubscriptionOutput: ...

deactivate_subscription

async def deactivate_subscription(
    *,
    domain_id: str,
    subscription_id: str,
) -> DeactivateSubscriptionOutput: ...

delete_domain

async def delete_domain(
    *,
    domain_id: str,
) -> DeleteDomainOutput: ...

get_domain

async def get_domain(
    *,
    domain_id: str,
) -> GetDomainOutput: ...

get_medical_scribe_listening_session

async def get_medical_scribe_listening_session(
    *,
    session_id: str,
    domain_id: str,
    subscription_id: str,
) -> GetMedicalScribeListeningSessionOutput: ...

get_patient_insights_job

async def get_patient_insights_job(
    *,
    domain_id: str,
    job_id: str,
) -> GetPatientInsightsJobOutput: ...

get_subscription

async def get_subscription(
    *,
    domain_id: str,
    subscription_id: str,
) -> GetSubscriptionOutput: ...

list_domains

async def list_domains(
    *,
    status: DomainStatus | None = None,
    max_results: int | None = None,
    next_token: str | None = None,
) -> ListDomainsOutput: ...

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

list_subscriptions

async def list_subscriptions(
    *,
    domain_id: str,
    max_results: int | None = None,
    next_token: str | None = None,
) -> ListSubscriptionsOutput: ...

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

list_tags_for_resource

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

start_medical_scribe_listening_session

async def start_medical_scribe_listening_session(
    *,
    session_id: str,
    domain_id: str,
    subscription_id: str,
    language_code: MedicalScribeLanguageCode,
    media_sample_rate_hertz: int,
    media_encoding: MedicalScribeMediaEncoding,
    input_stream: AsyncIterator[MedicalScribeInputStream] | None = None,
) -> StartMedicalScribeListeningSessionOutput: ...

start_patient_insights_job

async def start_patient_insights_job(
    *,
    domain_id: str,
    patient_context: PatientInsightsPatientContext,
    insights_context: InsightsContext,
    encounter_context: PatientInsightsEncounterContext,
    user_context: UserContext,
    input_data_config: InputDataConfig,
    output_data_config: OutputDataConfig,
    client_token: str | None = None,
) -> StartPatientInsightsJobOutput: ...

tag_resource

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

untag_resource

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

Links

• Service guide
• Repository
• Issues
• Releases

Licensed under Apache-2.0.