Connectors

Connectors combine AWS agent runtime integration with TAC channel management.

tac_aws.connectors

Connectors for AWS agent integrations with TAC.

Connectors combine agent runtime integration with channel management and conversation handling.

BedrockAgentCoreConnector

BedrockAgentCoreConnector(
    tac: TAC,
    runtime: RuntimeConfig | dict[str, Any],
    sms_config: SMSChannelConfig
    | dict[str, Any]
    | None = None,
    voice_config: VoiceChannelConfig
    | dict[str, Any]
    | None = None,
)

Connector for AWS Bedrock Agent Core with dual-runtime pattern.

Provides two runtime modes:

• HTTP invocation (required): For both voice and SMS channels
• WebSocket streaming (optional): For voice channel low-latency optimization (~50ms vs ~200ms)

Parameters:

Name	Type	Description	Default
tac	TAC	TAC instance for channel integration	required
runtime	RuntimeConfig | dict[str, Any]	Agent runtime configuration (RuntimeConfig or dict): http: Function to invoke agent via HTTP (required) Signature: (context, user_message, memory_context) -> InvokeAgentRuntimeResponseTypeDef Users control all invoke_agent_runtime() parameters websocket: Optional WebSocketConfig for voice optimization: factory: Async function to create WebSocket connection Signature: (context) -> WebSocketClientProtocol Called once per session for connection pooling payload_fn: Function to build WebSocket message payload Signature: (context, user_message, memory_context) -> dict[str, Any] Called every message - users control payload format	required
sms_config	SMSChannelConfig | dict[str, Any] | None	Optional SMS channel configuration (SMSChannelConfig or dict)	None
voice_config	VoiceChannelConfig | dict[str, Any] | None	Optional Voice channel configuration (VoiceChannelConfig or dict)	None

Attributes:

Name	Type	Description
voice		VoiceChannel instance for voice conversations
sms		SMSChannel instance for SMS conversations

Example

import boto3
import json
import websockets
from bedrock_agentcore.runtime import AgentCoreRuntimeClient
from tac import TAC, TACConfig
from tac.models.session import ConversationSession
from tac.channels.sms import SMSChannelConfig
from tac.channels.voice import VoiceChannelConfig
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
from websockets.client import WebSocketClientProtocol

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

# HTTP: boto3 client provides invoke_agent_runtime()
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
) -> dict:
    payload_data = {"prompt": user_message}
    if memory_context:
        payload_data["memory_context"] = memory_context

    payload = json.dumps(payload_data).encode("utf-8")

    return agentcore_http_client.invoke_agent_runtime(
        agentRuntimeArn=AGENT_ARN,
        runtimeSessionId=context.conversation_id,
        payload=payload,
    )

# WebSocket: AgentCoreRuntimeClient provides generate_ws_connection()
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[str, Any]:
    payload: dict[str, Any] = {"type": "prompt", "voicePrompt": user_message}
    if memory_context:
        payload["memoryContext"] = memory_context
    return payload

# Create connector
connector = BedrockAgentCoreConnector(
    tac=tac,
    runtime=RuntimeConfig(
        http=invoke_agent_http,  # Required: HTTP streaming for both channels
        websocket=WebSocketConfig(  # Optional: WebSocket optimization for voice
            factory=create_websocket,
            payload_fn=build_websocket_payload,
        ),
    ),
    voice_config=VoiceChannelConfig(
        session_manager=ThreadSafeSessionManager(),
        memory_mode="always",
    ),
    sms_config=SMSChannelConfig(memory_mode="always"),
)

# Use connector's channels for server
server = TACFastAPIServer(tac=tac, voice_channel=connector.voice, messaging_channels=[connector.sms])
server.start()

Initialize Bedrock Agent Core connector.

Parameters:

Name	Type	Description	Default
tac	TAC	TAC instance	required
runtime	RuntimeConfig | dict[str, Any]	Agent runtime configuration (RuntimeConfig or dict)	required
sms_config	SMSChannelConfig | dict[str, Any] | None	Optional SMS channel configuration	None
voice_config	VoiceChannelConfig | dict[str, Any] | None	Optional Voice channel configuration	None

BedrockConnector

BedrockConnector(
    tac: TAC,
    bedrock_client: AgentsforBedrockRuntimeClient
    | None = None,
    config: InvokeAgentRequestTypeDef
    | dict[str, Any]
    | None = None,
    invoke_fn: Callable[
        [ConversationSession, str, str | None],
        InvokeAgentResponseTypeDef,
    ]
    | None = None,
    sms_config: SMSChannelConfig
    | dict[str, Any]
    | None = None,
    voice_config: VoiceChannelConfig
    | dict[str, Any]
    | None = None,
)

Connector for AWS Bedrock Agents with multi-channel support.

Supports two usage patterns:

1. Simple config-based (recommended for most users)
2. Custom invoke function (for advanced use cases needing dynamic behavior)

Parameters:

Name	Type	Description	Default
tac	TAC	TAC instance for channel integration	required
bedrock_client	AgentsforBedrockRuntimeClient | None	AWS Bedrock Agent Runtime client (required if using config)	None
config	InvokeAgentRequestTypeDef | dict[str, Any] | None	Static configuration dict for invoke_agent() call (InvokeAgentRequestTypeDef). Required fields agentId, agentAliasId will be used. sessionId and inputText are auto-injected by connector. (required if using config pattern)	None
invoke_fn	Callable[[ConversationSession, str, str | None], InvokeAgentResponseTypeDef] | None	Custom function to invoke agent. Receives: context: ConversationSession with conversation_id, channel, etc. user_message: The user's message text memory_context: Optional memory context string (from TAC memory) Returns: InvokeAgentResponseTypeDef from client.invoke_agent() (required if not using config pattern)	None
sms_config	SMSChannelConfig | dict[str, Any] | None	Optional SMS channel configuration (SMSChannelConfig or dict)	None
voice_config	VoiceChannelConfig | dict[str, Any] | None	Optional Voice channel configuration (VoiceChannelConfig or dict)	None

Attributes:

Name	Type	Description
voice		VoiceChannel instance for voice conversations
sms		SMSChannel instance for SMS conversations

Example (Simple - Recommended):

import boto3
from tac import TAC, TACConfig
from tac.server import TACFastAPIServer
from tac_aws.connectors import BedrockConnector

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

# Simple config-based approach
connector = BedrockConnector(
    tac=tac,
    bedrock_client=client,
    config={
        "agentId": "AGENT123",
        "agentAliasId": "TSTALIASID",
        "enableTrace": False,  # Optional parameters
    }
)

server = TACFastAPIServer(tac=tac, voice_channel=connector.voice, sms_channel=connector.sms)
server.start()

Example (Advanced - Custom Logic):

import boto3
from tac import TAC, TACConfig
from tac.models.session import ConversationSession
from tac.server import TACFastAPIServer
from tac_aws.connectors import BedrockConnector

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

# Custom invoke function for dynamic behavior
def invoke_agent(
    context: ConversationSession,
    user_message: str,
    memory_context: str | None
):
    # Dynamic agent selection based on channel
    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="TSTALIASID",
        sessionId=context.conversation_id,
        inputText=full_message
    )

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

server = TACFastAPIServer(tac=tac, voice_channel=connector.voice, sms_channel=connector.sms)
server.start()

Initialize Bedrock Agent connector.

Parameters:

Name	Type	Description	Default
tac	TAC	TAC instance	required
bedrock_client	AgentsforBedrockRuntimeClient | None	AWS Bedrock Agent Runtime client (required if using config)	None
config	InvokeAgentRequestTypeDef | dict[str, Any] | None	Static invoke_agent config dict (required if using config pattern)	None
invoke_fn	Callable[[ConversationSession, str, str | None], InvokeAgentResponseTypeDef] | None	Custom invoke function (required if not using config pattern)	None
sms_config	SMSChannelConfig | dict[str, Any] | None	Optional SMS channel configuration	None
voice_config	VoiceChannelConfig | dict[str, Any] | None	Optional Voice channel configuration	None

Raises:

Type	Description
ValueError	If both invoke_fn and config are provided, or neither are provided

StrandsConnector

StrandsConnector(
    tac: TAC,
    agent_factory: Callable[[ConversationSession], Agent],
    sms_config: SMSChannelConfig
    | dict[str, Any]
    | None = None,
    voice_config: VoiceChannelConfig
    | dict[str, Any]
    | None = None,
)

Connector for AWS Strands SDK with multi-channel support.

Combines agent management with channel handling:

• Creates one Strands agent instance per conversation for proper isolation
• Manages Voice and SMS channels
• Handles memory injection and message routing

Parameters:

Name	Type	Description	Default
tac	TAC	TAC instance for channel integration	required
agent_factory	Callable[[ConversationSession], Agent]	Factory function that creates a new Agent instance. Receives ConversationSession context to enable SessionManager usage and context-aware agent configuration.	required
sms_config	SMSChannelConfig | dict[str, Any] | None	Optional SMS channel configuration (SMSChannelConfig or dict)	None
voice_config	VoiceChannelConfig | dict[str, Any] | None	Optional Voice channel configuration (VoiceChannelConfig or dict)	None

Attributes:

Name	Type	Description
voice		VoiceChannel instance for voice conversations
sms		SMSChannel instance for SMS conversations

Example

from tac import TAC, TACConfig
from tac.server import TACFastAPIServer
from tac.models.session import ConversationSession
from tac_aws.connectors import StrandsConnector
from strands import Agent
from strands.session.file import FileSessionManager

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

# Agent factory with context for SessionManager
def create_agent(context: ConversationSession) -> Agent:
    return Agent(
        model="amazon.nova-pro-v1:0",
        system_prompt="You are helpful.",
        session_manager=FileSessionManager(
            session_id=context.conversation_id,
            base_path="./sessions"
        )
    )

# Create connector with agent factory
connector = StrandsConnector(tac=tac, agent_factory=create_agent)

# Use connector's channels for server
server = TACFastAPIServer(tac=tac, voice_channel=connector.voice, sms_channel=connector.sms)
server.start()

Initialize Strands connector with agent factory and channel configs.

Parameters:

Name	Type	Description	Default
tac	TAC	TAC instance	required
agent_factory	Callable[[ConversationSession], Agent]	Factory function that creates a new Agent instance. Receives ConversationSession context as parameter.	required
sms_config	SMSChannelConfig | dict[str, Any] | None	Optional SMS channel configuration	None
voice_config	VoiceChannelConfig | dict[str, Any] | None	Optional Voice channel configuration	None

AgentCore Runtime Configuration

Configuration objects for BedrockAgentCoreConnector.

tac_aws.connectors.bedrock_agentcore.config

Configuration schemas for AgentCore connector.

WebSocketConfig

Bases: BaseModel

WebSocket configuration for voice channel optimization.

RuntimeConfig

Bases: BaseModel

Runtime configuration for AgentCore connector.

validate_websocket classmethod

validate_websocket(
    v: dict[str, Any] | WebSocketConfig | None,
) -> WebSocketConfig | None

Convert dict to WebSocketConfig if needed.