AWS connectors

This guide shows how to connect AWS agents to Twilio communication channels using the 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.

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)
An AWS account with access to Amazon 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. Recommended for most use cases.	`strands`
`BedrockConnector`	Use a managed Bedrock Agent with an agent ID and alias.	`bedrock`
`BedrockAgentCoreConnector`	Use Bedrock AgentCore 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

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. Make sure your AWS credentials are configured through the standard AWS credential chain (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.

1from dotenv import load_dotenv
2from strands import Agent
3from tac import TAC
4from tac.core.config import TACConfig
5from tac.channels.voice import VoiceChannelConfig
6from tac.channels.sms import SMSChannelConfig
7from tac.models.session import ConversationSession
8from tac.server import TACFastAPIServer
9
10from tac_aws.connectors import StrandsConnector
11
12load_dotenv()
13
14tac = TAC(config=TACConfig.from_env())
15
16
17def create_agent(context: ConversationSession) -> Agent:
18    return Agent(
19        model="amazon.nova-pro-v1:0",
20        system_prompt="You are a helpful customer service agent. Be concise and friendly.",
21    )
22
23
24connector = StrandsConnector(
25    tac=tac,
26    agent_factory=create_agent,
27    voice_config=VoiceChannelConfig(memory_mode="always"),
28    sms_config=SMSChannelConfig(memory_mode="always"),
29)
30
31server = TACFastAPIServer(
32    tac=tac,
33    voice_channel=connector.voice,
34    messaging_channels=[connector.sms],
35)
36
37if __name__ == "__main__":
38    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

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

Configure environment variables

Add the following to your .env file alongside your existing TAC credentials.

1BEDROCK_AGENT_ID=your_agent_id
2BEDROCK_AGENT_ALIAS_ID=your_agent_alias_id
3AWS_REGION=us-east-1

Set up the server

1import boto3
2from dotenv import load_dotenv
3from tac import TAC
4from tac.core.config import TACConfig
5from tac.server import TACFastAPIServer
6
7from tac_aws.connectors import BedrockConnector
8
9load_dotenv()
10
11tac = TAC(config=TACConfig.from_env())
12client = boto3.client("bedrock-agent-runtime", region_name="us-east-1")
13
14connector = BedrockConnector(
15    tac=tac,
16    bedrock_client=client,
17    config={
18        "agentId": "YOUR_AGENT_ID",
19        "agentAliasId": "YOUR_AGENT_ALIAS_ID",
20    },
21)
22
23server = TACFastAPIServer(
24    tac=tac,
25    voice_channel=connector.voice,
26    messaging_channels=[connector.sms],
27)
28
29if __name__ == "__main__":
30    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.

1from tac.models.session import ConversationSession
2
3def invoke_agent(
4    context: ConversationSession,
5    user_message: str,
6    memory_context: str | None,
7):
8    agent_id = "VOICE_AGENT" if context.channel == "voice" else "SMS_AGENT"
9
10    full_message = user_message
11    if memory_context:
12        full_message = f"{memory_context}\n\nUser: {user_message}"
13
14    return client.invoke_agent(
15        agentId=agent_id,
16        agentAliasId="YOUR_AGENT_ALIAS_ID",
17        sessionId=context.conversation_id,
18        inputText=full_message,
19    )
20
21connector = BedrockConnector(tac=tac, invoke_fn=invoke_agent)

Bedrock AgentCore connector

Bedrock AgentCore supports a dual-runtime pattern: HTTP for SMS and a WebSocket option for low-latency voice streaming.

Install dependencies

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

Configure environment variables

Add the following to your .env file alongside your existing TAC credentials.

1BEDROCK_AGENTCORE_AGENT_ARN=arn:aws:bedrock-agentcore:us-east-1:123456789012:runtime/agent-xxx
2AWS_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.

1import json
2import boto3
3from dotenv import load_dotenv
4from tac import TAC
5from tac.core.config import TACConfig
6from tac.channels.voice import VoiceChannelConfig
7from tac.models.session import ConversationSession
8from tac.server import TACFastAPIServer
9from tac.session import ThreadSafeSessionManager
10
11from tac_aws.connectors import BedrockAgentCoreConnector
12from tac_aws.connectors.bedrock_agentcore.config import RuntimeConfig, WebSocketConfig
13
14load_dotenv()
15
16tac = TAC(config=TACConfig.from_env())
17AGENT_ARN = "arn:aws:bedrock-agentcore:us-east-1:123456789012:runtime/agent-xxx"
18
19# HTTP invoke function (required for both Voice and SMS)
20agentcore_http_client = boto3.client("bedrock-agentcore", region_name="us-east-1")
21
22
23def invoke_agent_http(
24    context: ConversationSession,
25    user_message: str,
26    memory_context: str | None,
27):
28    payload_data = {"prompt": user_message}
29    if memory_context:
30        payload_data["memory_context"] = memory_context
31
32    return agentcore_http_client.invoke_agent_runtime(
33        agentRuntimeArn=AGENT_ARN,
34        runtimeSessionId=context.conversation_id,
35        payload=json.dumps(payload_data).encode("utf-8"),
36    )
37
38
39connector = BedrockAgentCoreConnector(
40    tac=tac,
41    runtime=RuntimeConfig(http=invoke_agent_http),
42    voice_config=VoiceChannelConfig(
43        session_manager=ThreadSafeSessionManager(),
44        memory_mode="always",
45    ),
46)
47
48server = TACFastAPIServer(
49    tac=tac,
50    voice_channel=connector.voice,
51    messaging_channels=[connector.sms],
52)
53
54if __name__ == "__main__":
55    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.

1import websockets
2from bedrock_agentcore.runtime import AgentCoreRuntimeClient
3from websockets.client import WebSocketClientProtocol
4
5agentcore_client = AgentCoreRuntimeClient(region="us-east-1")
6
7
8async def create_websocket(context: ConversationSession) -> WebSocketClientProtocol:
9    ws_url, headers = agentcore_client.generate_ws_connection(
10        runtime_arn=AGENT_ARN,
11        session_id=context.conversation_id,
12    )
13    return await websockets.connect(ws_url, additional_headers=headers)
14
15
16def build_websocket_payload(
17    context: ConversationSession,
18    user_message: str,
19    memory_context: str | None,
20) -> dict:
21    payload = {"type": "prompt", "voicePrompt": user_message}
22    if memory_context:
23        payload["memoryContext"] = memory_context
24    return payload
25
26
27connector = BedrockAgentCoreConnector(
28    tac=tac,
29    runtime=RuntimeConfig(
30        http=invoke_agent_http,
31        websocket=WebSocketConfig(
32            factory=create_websocket,
33            payload_fn=build_websocket_payload,
34        ),
35    ),
36    voice_config=VoiceChannelConfig(
37        session_manager=ThreadSafeSessionManager(),
38        memory_mode="always",
39    ),
40)

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.

Clone the repository and install dependencies:

1git clone https://github.com/twilio/twilio-agent-connect-aws.git
2cd twilio-agent-connect-aws
3make sync
4make dev-setup

Create a .env file in the examples directory:

cp getting_started/examples/.env.example getting_started/examples/.env

Fill in your TAC credentials and the AWS variables for the connector you want to run:
StrandsBedrock AgentAgentCore
Strands uses the standard AWS credential chain — no Bedrock agent IDs needed.
AWS_REGION=us-east-1
1
BEDROCK_AGENT_ID=your_agent_id
2
BEDROCK_AGENT_ALIAS_ID=your_agent_alias_id
3
AWS_REGION=us-east-1
1
BEDROCK_AGENTCORE_AGENT_ARN=arn:aws:bedrock-agentcore:us-east-1:123456789012:runtime/agent-xxx
2
AWS_REGION=us-east-1

From the repository root, start the example server:

StrandsBedrock AgentAgentCore

uv run getting_started/examples/strands_agents.py

uv run getting_started/examples/bedrock_agents.py

uv run getting_started/examples/bedrock_agentcore_agents.py

Expose your local server with ngrok 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 and Configure Twilio webhooks.

Test the integration

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

Next steps

Deploy to AWS Fargate: Deploy your TAC application to production on AWS.
Add TAC to your agent: Add custom tools and knowledge search.
Troubleshooting: Common issues and solutions.