# AWS connectors

This guide shows how to connect AWS agents to Twilio communication channels using the [`twilio-agent-connect-aws`](https://github.com/twilio/twilio-agent-connect-aws) package. The package provides connectors that handle channel routing and memory context. This guide covers three connectors: Strands (recommended), Bedrock Agent, and Bedrock AgentCore.

When you're ready to deploy to production, see [Deploy to AWS Fargate](/docs/conversations/agent-connect/integrations/aws/deploy).

## Prerequisites

Before you begin, make sure you have:

* TAC SDK installed and configured with your Twilio credentials and a Conversation Configuration (see [Add TAC to your agent prerequisites](/docs/conversations/agent-connect/build-with-tac#prerequisites))
* An AWS account with access to [Amazon Bedrock](https://aws.amazon.com/bedrock/)
* A Bedrock foundation model activated in your AWS region (for example, Amazon Nova, Anthropic Claude, or Meta Llama)
* AWS credentials configured locally (`AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_REGION`)
* Python 3.10+

## Choose a connector

The `twilio-agent-connect-aws` package provides connectors for different AWS agent runtimes.

| Connector                   | Use case                                                                                                                       | Install extra |
| --------------------------- | ------------------------------------------------------------------------------------------------------------------------------ | ------------- |
| `StrandsConnector`          | Build agents with the [AWS Strands Agents SDK](https://github.com/strands-agents/sdk-python). Recommended for most use cases.  | `strands`     |
| `BedrockConnector`          | Use a managed [Bedrock Agent](https://docs.aws.amazon.com/bedrock/latest/userguide/agents.html) with an agent ID and alias.    | `bedrock`     |
| `BedrockAgentCoreConnector` | Use [Bedrock AgentCore](https://docs.aws.amazon.com/bedrock/latest/userguide/agentcore.html) with HTTP and WebSocket runtimes. | `agentcore`   |

## Strands connector

The Strands connector is the simplest way to connect a Bedrock-powered agent to Twilio channels.

### Install dependencies

```bash
pip install twilio-agent-connect-aws[strands,server]
```

The Strands SDK manages Bedrock calls internally, so no additional AWS environment variables are needed beyond your existing [TAC credentials](/docs/conversations/agent-connect/build-with-tac#configure-environment-variables). Make sure your AWS credentials are configured through the [standard AWS credential chain](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-files.html) (environment variables, `~/.aws/credentials`, or IAM role).

### Set up the server

The `StrandsConnector` takes an agent factory function that creates a new Strands `Agent` for each conversation. The connector handles memory context, channel routing, and conversation cleanup.

```python
from dotenv import load_dotenv
from strands import Agent
from tac import TAC
from tac.core.config import TACConfig
from tac.channels.voice import VoiceChannelConfig
from tac.channels.sms import SMSChannelConfig
from tac.models.session import ConversationSession
from tac.server import TACFastAPIServer

from tac_aws.connectors import StrandsConnector

load_dotenv()

tac = TAC(config=TACConfig.from_env())


def create_agent(context: ConversationSession) -> Agent:
    return Agent(
        model="amazon.nova-pro-v1:0",
        system_prompt="You are a helpful customer service agent. Be concise and friendly.",
    )


connector = StrandsConnector(
    tac=tac,
    agent_factory=create_agent,
    voice_config=VoiceChannelConfig(memory_mode="always"),
    sms_config=SMSChannelConfig(memory_mode="always"),
)

server = TACFastAPIServer(
    tac=tac,
    voice_channel=connector.voice,
    messaging_channels=[connector.sms],
)

if __name__ == "__main__":
    server.start()
```

The connector creates one agent per conversation and automatically:

* Retrieves memory context from Conversation Memory and includes it in the first message (when `memory_mode` is set to `"always"` or `"once"`)
* Routes responses to the correct channel (Voice or SMS)
* Cleans up agent resources when the conversation ends

To disable memory retrieval, omit `voice_config` and `sms_config` or set `memory_mode` to `"never"`.

## Bedrock Agent connector

If you have a managed Bedrock Agent, use the `BedrockConnector` to connect it to Twilio channels.

### Install dependencies

```bash
pip install twilio-agent-connect-aws[bedrock,server]
```

### Configure environment variables

Add the following to your `.env` file alongside your existing [TAC credentials](/docs/conversations/agent-connect/build-with-tac#configure-environment-variables).

```text
BEDROCK_AGENT_ID=your_agent_id
BEDROCK_AGENT_ALIAS_ID=your_agent_alias_id
AWS_REGION=us-east-1
```

### Set up the server

```python
import boto3
from dotenv import load_dotenv
from tac import TAC
from tac.core.config import TACConfig
from tac.server import TACFastAPIServer

from tac_aws.connectors import BedrockConnector

load_dotenv()

tac = TAC(config=TACConfig.from_env())
client = boto3.client("bedrock-agent-runtime", region_name="us-east-1")

connector = BedrockConnector(
    tac=tac,
    bedrock_client=client,
    config={
        "agentId": "YOUR_AGENT_ID",
        "agentAliasId": "YOUR_AGENT_ALIAS_ID",
    },
)

server = TACFastAPIServer(
    tac=tac,
    voice_channel=connector.voice,
    messaging_channels=[connector.sms],
)

if __name__ == "__main__":
    server.start()
```

The connector automatically passes `sessionId` and `inputText` to each `invoke_agent` call. If Conversation Memory is configured, memory context is prepended to the user's message.

For advanced use cases like dynamic agent selection based on channel, you can provide a custom invoke function instead of a static config.

```python
from tac.models.session import ConversationSession

def invoke_agent(
    context: ConversationSession,
    user_message: str,
    memory_context: str | None,
):
    agent_id = "VOICE_AGENT" if context.channel == "voice" else "SMS_AGENT"

    full_message = user_message
    if memory_context:
        full_message = f"{memory_context}\n\nUser: {user_message}"

    return client.invoke_agent(
        agentId=agent_id,
        agentAliasId="YOUR_AGENT_ALIAS_ID",
        sessionId=context.conversation_id,
        inputText=full_message,
    )

connector = BedrockConnector(tac=tac, invoke_fn=invoke_agent)
```

## Bedrock AgentCore connector

[Bedrock AgentCore](https://docs.aws.amazon.com/bedrock/latest/userguide/agentcore.html) supports a dual-runtime pattern: HTTP for SMS and a WebSocket option for low-latency voice streaming.

### Install dependencies

```bash
pip install twilio-agent-connect-aws[agentcore,server]
```

### Configure environment variables

Add the following to your `.env` file alongside your existing [TAC credentials](/docs/conversations/agent-connect/build-with-tac#configure-environment-variables).

```text
BEDROCK_AGENTCORE_AGENT_ARN=arn:aws:bedrock-agentcore:us-east-1:123456789012:runtime/agent-xxx
AWS_REGION=us-east-1
```

### Set up the server

The `BedrockAgentCoreConnector` requires an HTTP invoke function. You can optionally add a WebSocket configuration for voice optimization.

```python
import json
import boto3
from dotenv import load_dotenv
from tac import TAC
from tac.core.config import TACConfig
from tac.channels.voice import VoiceChannelConfig
from tac.models.session import ConversationSession
from tac.server import TACFastAPIServer
from tac.session import ThreadSafeSessionManager

from tac_aws.connectors import BedrockAgentCoreConnector
from tac_aws.connectors.bedrock_agentcore.config import RuntimeConfig, WebSocketConfig

load_dotenv()

tac = TAC(config=TACConfig.from_env())
AGENT_ARN = "arn:aws:bedrock-agentcore:us-east-1:123456789012:runtime/agent-xxx"

# HTTP invoke function (required for both Voice and SMS)
agentcore_http_client = boto3.client("bedrock-agentcore", region_name="us-east-1")


def invoke_agent_http(
    context: ConversationSession,
    user_message: str,
    memory_context: str | None,
):
    payload_data = {"prompt": user_message}
    if memory_context:
        payload_data["memory_context"] = memory_context

    return agentcore_http_client.invoke_agent_runtime(
        agentRuntimeArn=AGENT_ARN,
        runtimeSessionId=context.conversation_id,
        payload=json.dumps(payload_data).encode("utf-8"),
    )


connector = BedrockAgentCoreConnector(
    tac=tac,
    runtime=RuntimeConfig(http=invoke_agent_http),
    voice_config=VoiceChannelConfig(
        session_manager=ThreadSafeSessionManager(),
        memory_mode="always",
    ),
)

server = TACFastAPIServer(
    tac=tac,
    voice_channel=connector.voice,
    messaging_channels=[connector.sms],
)

if __name__ == "__main__":
    server.start()
```

### Add WebSocket for voice optimization

For lower latency on the Voice channel (approximately 50ms vs 200ms), add a WebSocket configuration. The connector uses WebSocket for voice and falls back to HTTP for SMS.

```python
import websockets
from bedrock_agentcore.runtime import AgentCoreRuntimeClient
from websockets.client import WebSocketClientProtocol

agentcore_client = AgentCoreRuntimeClient(region="us-east-1")


async def create_websocket(context: ConversationSession) -> WebSocketClientProtocol:
    ws_url, headers = agentcore_client.generate_ws_connection(
        runtime_arn=AGENT_ARN,
        session_id=context.conversation_id,
    )
    return await websockets.connect(ws_url, additional_headers=headers)


def build_websocket_payload(
    context: ConversationSession,
    user_message: str,
    memory_context: str | None,
) -> dict:
    payload = {"type": "prompt", "voicePrompt": user_message}
    if memory_context:
        payload["memoryContext"] = memory_context
    return payload


connector = BedrockAgentCoreConnector(
    tac=tac,
    runtime=RuntimeConfig(
        http=invoke_agent_http,
        websocket=WebSocketConfig(
            factory=create_websocket,
            payload_fn=build_websocket_payload,
        ),
    ),
    voice_config=VoiceChannelConfig(
        session_manager=ThreadSafeSessionManager(),
        memory_mode="always",
    ),
)
```

The connector pools WebSocket connections per conversation and automatically cleans them up when conversations end.

## Run the example

To start from a working example instead of writing the server code yourself, clone the `twilio-agent-connect-aws` repository and run one of the included examples.

1. Clone the repository and install dependencies:

   ```bash
   git clone https://github.com/twilio/twilio-agent-connect-aws.git
   cd twilio-agent-connect-aws
   make sync
   make dev-setup
   ```
2. Create a `.env` file in the examples directory:

   ```bash
   cp getting_started/examples/.env.example getting_started/examples/.env
   ```
3. Fill in your [TAC credentials](/docs/conversations/agent-connect/build-with-tac#configure-environment-variables) and the AWS variables for the connector you want to run:

   ## Strands

   Strands uses the standard AWS credential chain — no Bedrock agent IDs needed.

   ```text
   AWS_REGION=us-east-1
   ```

   ## Bedrock Agent

   ```text
   BEDROCK_AGENT_ID=your_agent_id
   BEDROCK_AGENT_ALIAS_ID=your_agent_alias_id
   AWS_REGION=us-east-1
   ```

   ## AgentCore

   ```text
   BEDROCK_AGENTCORE_AGENT_ARN=arn:aws:bedrock-agentcore:us-east-1:123456789012:runtime/agent-xxx
   AWS_REGION=us-east-1
   ```
4. From the repository root, start the example server:

   ## Strands

   ```bash
   uv run getting_started/examples/strands_agents.py
   ```

   ## Bedrock Agent

   ```bash
   uv run getting_started/examples/bedrock_agents.py
   ```

   ## AgentCore

   ```bash
   uv run getting_started/examples/bedrock_agentcore_agents.py
   ```
5. Expose your local server with [ngrok](https://ngrok.com/) so Twilio can reach it, then configure your Twilio webhooks to point at the ngrok domain. For detailed steps, see [Expose your local server with ngrok](/docs/conversations/agent-connect/quickstart#expose-your-local-server-with-ngrok) and [Configure Twilio webhooks](/docs/conversations/agent-connect/quickstart#configure-twilio-webhooks).

## Test the integration

1. Start your server and make sure ngrok is running.
2. Send an SMS to your Twilio phone number and verify that you receive a response.
3. Call your Twilio phone number and verify that the voice channel works.
4. If you configured Conversation Memory, verify that responses include personalized context from the customer's profile.

## Next steps

* [Deploy to AWS Fargate](/docs/conversations/agent-connect/integrations/aws/deploy): Deploy your TAC application to production on AWS.
* [Add TAC to your agent](/docs/conversations/agent-connect/build-with-tac): Add custom tools and knowledge search.
* [Troubleshooting](/docs/conversations/agent-connect/troubleshooting): Common issues and solutions.