Strands Connect

A reusable Amazon Connect native-voice extension for Strands + Amazon Bedrock AgentCore. Bring your own voice model, system prompt, and business tools. The adapter handles Connect A2A audio, contact-scoped tools, automatic contact attribute persistence, tool sounds, transcripts, and handback.

Early implementation, v0.1.0. Independent community extension; not an AWS, Strands, OpenAI, or Google product. The realtime Strands API is experimental, so tested versions are pinned. See verification and limitations.

flowchart LR
    Caller <--> Connect[Amazon Connect]
    Connect <-->|A2A voice / bearer auth| Bridge[Thin ingress bridge]
    Bridge <-->|SigV4 WebSocket| Runtime[AgentCore Runtime]
    subgraph Runtime
        Adapter[ConnectSession + hooks] <--> Agent[Strands BidiAgent]
        Agent <--> Tools[Your business tools]
    end
    Agent <--> Provider[OpenAI Realtime / Gemini Live / Nova Sonic]
    Adapter --> Attributes[Connect contact attributes and details]
    Adapter --> Traces[A2A transcripts and tool traces]

Python 3.12+. Until a PyPI release exists, install directly from this repository:

pip install 'strands-connect[openai,agentcore] @ git+https://github.com/caylent/strands-connect.git@main'
# Or select gemini instead of openai. Pin a commit for reproducible applications.

No microphone or PyAudio dependency is required on the server. Provider imports are optional and lazy.

• Connect + OpenAI Realtime SDK
• Connect + Gemini Live

Both run the same synthetic order lookup and contact lifecycle on AgentCore. They differ only in provider configuration. See AgentCore and Connect setup for the ingress bridge, IAM, registration, recording, and validation steps. There are no account IDs, secrets, or pre-existing demo resources required by the source code.

from strands import tool
from strands_connect import ConnectSession, ContactPolicy
from strands_connect.contact import SESSION_ATTRIBUTES
from strands_connect.providers import make_model


@tool
async def lookup_order(order_id: str) -> dict:
    """Look up an order in your application."""
    return {"id": order_id, "status": "Shipped"}  # Replace with your API.


session = ConnectSession(
    socket,  # Async iterator of text frames, with send(), close(), and headers.
    instance_arn=your_instance_arn,
    connect_client=your_boto3_connect_client,
    model_factory=lambda: make_model("openai", "gpt-realtime", api_key=your_key),
    provider="openai",
    system_prompt="Help with order status; use tools for factual answers.",
    tools_factory=lambda session: [lookup_order],
    contact_policy=ContactPolicy(
        allowed_attributes=SESSION_ATTRIBUTES | {"OrderId", "OrderStatus"},
    ),
    result_mappers={
        "lookup_order": lambda result: {"OrderId": result["id"], "OrderStatus": result["status"]},
    },
)
await session.run()

To host your configured session in AgentCore, use the library's app factory:

from strands_connect.agentcore import create_app

# Your factory receives the socket and a trusted runtime session ID.
app = create_app(your_session_factory)
app.run()

The AfterToolCallEvent hook saves and reads back mapped results before the model receives the result. Tools do not need to remember a separate logging call. A failed save becomes an error result rather than an unverified success. Never automatically repeat a non-idempotent business action to recover a logging failure.

• agentcore.create_app: reusable /ws entrypoint for your own session factory.
• ConnectSession: one trusted contact, one model instance, one Strands BidiAgent.
• ConnectContactHooks: public async Strands tool hooks, installed through hooks=[...].
• contact_tools: standard @tool functions for approved attributes and completion/escalation; optional name/description updates.
• ContactStore: bounded writer queue, API readback, explicit failure and shutdown semantics.
• model_factory: any audio-capable BidiModel with PCM16 mono at 8/16/24 kHz.

This follows Strands' extension template and Bidi hooks patterns. BidiAgent 1.56 exposes hooks and tools; its constructor does not register standard Agent(plugins=[...]) plugins. Consequently this package uses those supported APIs rather than an ignored plugins parameter.

Contact persistence distinguishes custom attributes, contact details, transcript/tool traces, and audio recordings. A CTR is a generated contact record, not an arbitrary document to patch. The adapter uses supported Connect APIs and A2A events; it does not rewrite CTR exports.

See the provider guide for OpenAI SDK behavior, Gemini compatibility fixes, Nova Sonic, and custom providers. See CONTRIBUTING for local checks and extension work.

uv sync --all-extras --group dev
uv run pytest
uv run ruff check .
uv build

Apache License 2.0.