Skip to content

Channels

tac.channels

Communication channels for the Twilio Agent Connect.

BaseChannel

BaseChannel(
    tac: TAC,
    memory_mode: MemoryMode = "never",
    dedup_capacity: int = 10000,
)

Bases: ABC

Abstract base class for TAC channels.

Channels handle protocol-specific webhook processing and response delivery for different communication channels (SMS, Voice, etc.).

This class provides common conversation lifecycle management that is shared across all channel types.

Initialize base channel.

Parameters:

Name Type Description Default
tac TAC

TAC instance for memory/context operations

required
memory_mode MemoryMode

Memory retrieval mode. Default is "never". - "always": Retrieve memory for every message with the query string - "once": Retrieve memory once at conversation start with empty query and cache it. Cache is invalidated when conversation becomes INACTIVE. - "never": Skip memory retrieval

'never'
dedup_capacity int

Maximum number of idempotency tokens to track for webhook deduplication. Default 10000. Must be positive.

10000

process_webhook abstractmethod async

process_webhook(
    webhook_data: dict[str, Any],
    idempotency_token: str | None = None,
) -> None

Process incoming webhook event from Twilio.

This method should: 1. Parse and validate webhook data 2. Handle conversation lifecycle (start, message, end) 3. Trigger memory retrieval via TAC 4. Invoke registered callbacks

Parameters:

Name Type Description Default
webhook_data dict[str, Any]

Raw webhook event data from Twilio

required
idempotency_token str | None

Optional Twilio idempotency token from request headers

None

send_response abstractmethod async

send_response(
    conversation_id: str,
    response: str
    | AsyncGenerator[str | dict[str, Any], None],
    role: str | None = None,
) -> None

Send response back through the channel.

Supports both simple string responses and streaming via async generators.

Parameters:

Name Type Description Default
conversation_id str

Conversation ID to send response to

required
response str | AsyncGenerator[str | dict[str, Any], None]

Message content (string) or async generator for streaming

required
role str | None

Optional message role (e.g., 'assistant', 'user', 'system')

None

get_channel_name abstractmethod

get_channel_name() -> str

Get the channel name identifier.

Returns:

Type Description
str

Channel name (e.g., 'sms', 'voice')

get_channel_type_upper

get_channel_type_upper() -> str

Get uppercase channel type for webhook filtering.

Returns:

Type Description
str

Uppercase channel type (e.g., 'SMS', 'VOICE')

ChatChannel

ChatChannel(
    tac: TAC,
    config: ChatChannelConfig
    | dict[str, Any]
    | None = None,
)

Bases: MessagingChannel

Chat Channel for handling web chat conversations.

Uses identity-based addressing instead of phone numbers. Automatically creates AI_AGENT participant if needed (lazy creation) and manages conversation lifecycle through Conversation Orchestrator webhooks.

send_response async

send_response(
    conversation_id: str,
    response: str
    | AsyncGenerator[str | dict[str, Any], None],
    role: str | None = None,
) -> None

Send chat response using the Conversation Orchestrator Send API.

Reads the agent and customer participant ids stashed on the session by inbound reconciliation or outbound initiation. Missing ids are a misuse — send_response is only expected to be called after an inbound webhook (COMMUNICATION_CREATED → reconcile) or after initiate_outbound_conversation, both of which populate the session.

Parameters:

Name Type Description Default
conversation_id str

Conversation ID to send response to

required
response str | AsyncGenerator[str | dict[str, Any], None]

Message content (must be string for Chat)

required
role str | None

Optional message role (not used in Chat channel)

None

Raises:

Type Description
TypeError

If response is not a string

RuntimeError

If the session, channel_id, or participant ids are missing

initiate_outbound_conversation async

initiate_outbound_conversation(
    options: InitiateChatConversationOptions,
) -> InitiateConversationResult

Initiate an outbound Chat conversation.

Creates a conversation via Conversation Orchestrator with inline participants, then sends the initial message via the Actions API. If an active conversation with the same addresses already exists (group-by dedup), CO returns 409 and the existing conversation is reused.

ChatChannelConfig

Bases: MessagingChannelConfig

Configuration for Chat channel.

Attributes:

Name Type Description
agent_address str

Chat agent identity string used to identify the bot's messages.

MessagingChannel

MessagingChannel(
    tac: TAC,
    dedup_capacity: int = 10000,
    memory_mode: MemoryMode = "never",
)

Bases: BaseChannel

Abstract base class for messaging channels (SMS, RCS, WhatsApp, Chat).

Provides shared webhook processing logic for channels that use Conversation Orchestrator webhooks with COMMUNICATION_CREATED and CONVERSATION_UPDATED event types.

Subclasses must implement: - is_default_agent_address(): Fast-path check for the channel's default agent address - get_channel_type_upper(): Return uppercase channel type ("SMS", "RCS", "WHATSAPP", "CHAT") - get_agent_address(conversation_id): Return the agent's ParticipantAddress for a conversation - send_response(): Send messages back through the channel - get_channel_name(): Return lowercase channel name ("sms", "rcs", "whatsapp", "chat")

Subclass class attributes: - reconcile_customer_type: If True, reconciliation will also promote a channel-matching UNKNOWN participant (not owning the agent address) to CUSTOMER. Set False for channels where the customer is identified author-driven (e.g. chat).

is_default_agent_address abstractmethod

is_default_agent_address(author_address: str) -> bool

Fast-path check: is the author address this channel's default agent address?

For example, config.phone_number for SMS, config.rcs_sender_id for RCS, config.whatsapp_number for WhatsApp, agent_address for Chat.

Parameters:

Name Type Description Default
author_address str

The address of the message author

required

Returns:

Type Description
bool

True if the address matches the channel's default agent address

get_channel_type_upper abstractmethod

get_channel_type_upper() -> str

Return the uppercase channel type for webhook filtering.

Returns:

Type Description
str

Channel type string (e.g., "SMS", "CHAT")

get_agent_address abstractmethod

get_agent_address(
    conversation_id: str,
) -> ParticipantAddress

Return the agent-side ParticipantAddress for this conversation.

Used by _reconcile_participants to identify which participant (by channel + address) represents the agent. May read from session state (e.g. chat's per-conversation channelId) to build the address.

process_webhook async

process_webhook(
    webhook_data: dict[str, Any],
    idempotency_token: str | None = None,
) -> None

Process messaging channel webhook event and manage conversation lifecycle.

Handles: - COMMUNICATION_CREATED: Process incoming messages from customers - CONVERSATION_UPDATED: Clean up when conversation is closed

Note: Conversation tracking uses instance-local memory. In multi-instance deployments, webhooks may route to a different instance, preventing cleanup. See CLAUDE.md for horizontal scaling considerations.

Parameters:

Name Type Description Default
webhook_data dict[str, Any]

Raw webhook event data from Twilio

required
idempotency_token str | None

Optional Twilio idempotency token from request headers

None

MessagingChannelConfig

Bases: BaseModel

Base configuration for messaging channels (SMS, RCS, WhatsApp, Chat).

Attributes:

Name Type Description
dedup_capacity int

Maximum number of idempotency tokens to track. Default 10000 is suitable for most applications. Uses Twilio's i-twilio-idempotency-token header for deduplication.

memory_mode MemoryMode

Memory retrieval mode. Default is "never". - "always": Retrieve memory for every message with the query string - "once": Retrieve memory once at conversation start with empty query and cache it. Cache is invalidated when conversation becomes INACTIVE and is fetched again the next time a message triggers memory retrieval after the conversation becomes ACTIVE. - "never": Skip memory retrieval

RCSChannel

RCSChannel(
    tac: TAC,
    config: RCSChannelConfig | dict[str, Any] | None = None,
)

Bases: MessagingChannel

RCS Channel for handling RCS-based conversations.

Inherits shared messaging channel webhook processing from MessagingChannel and provides RCS-specific message sending and filtering.

RCS uses RCS Sender IDs configured in TACConfig (via TWILIO_RCS_SENDER_ID).

is_default_agent_address

is_default_agent_address(author_address: str) -> bool

Check if the author address matches the configured RCS sender ID.

get_agent_address

get_agent_address(
    conversation_id: str,
) -> ParticipantAddress

Get the agent's participant address for this conversation.

send_response async

send_response(
    conversation_id: str,
    response: str
    | AsyncGenerator[str | dict[str, Any], None],
    role: str | None = None,
) -> None

Send RCS response using the Conversation Orchestrator Send API.

Reads the agent and customer participant ids stashed on the session by inbound reconciliation or outbound initiation. Missing ids are a misuse — send_response is only expected to be called after an inbound webhook (COMMUNICATION_CREATED → reconcile) or after initiate_outbound_conversation, both of which populate the session.

Parameters:

Name Type Description Default
conversation_id str

Conversation ID to send response to

required
response str | AsyncGenerator[str | dict[str, Any], None]

Message content (must be string for RCS)

required
role str | None

Optional message role (not used in RCS channel)

None

Raises:

Type Description
TypeError

If response is not a string

RuntimeError

If the session or participant ids are missing

initiate_outbound_conversation async

initiate_outbound_conversation(
    options: InitiateMessagingConversationOptions,
) -> InitiateConversationResult

Initiate an outbound RCS conversation.

Creates a conversation via Conversation Orchestrator with inline participants, then sends the initial message via the Actions API. Uses the RCS sender ID from TACConfig as the from address. If an active conversation with the same addresses already exists (group-by dedup), CO returns 409 and the existing conversation is reused.

Parameters:

Name Type Description Default
options InitiateMessagingConversationOptions

Conversation initiation options (to address and message)

required

Returns:

Type Description
InitiateConversationResult

InitiateConversationResult with conversation_id and session

Raises:

Type Description
RuntimeError

If rcs_sender_id is not configured

RCSChannelConfig

Bases: MessagingChannelConfig

Configuration for RCS channel.

Inherits dedup_capacity and memory_mode from MessagingChannelConfig.

SMSChannel

SMSChannel(
    tac: TAC,
    config: SMSChannelConfig | dict[str, Any] | None = None,
)

Bases: MessagingChannel

SMS Channel for handling SMS-based conversations.

Inherits shared messaging channel webhook processing from MessagingChannel and provides SMS-specific message sending and filtering.

send_response async

send_response(
    conversation_id: str,
    response: str
    | AsyncGenerator[str | dict[str, Any], None],
    role: str | None = None,
) -> None

Send SMS response using the Conversation Orchestrator Send API.

Reads the agent and customer participant ids stashed on the session by inbound reconciliation or outbound initiation. Missing ids are a misuse — send_response is only expected to be called after an inbound webhook (COMMUNICATION_CREATED → reconcile) or after initiate_outbound_conversation, both of which populate the session.

Parameters:

Name Type Description Default
conversation_id str

Conversation ID to send response to

required
response str | AsyncGenerator[str | dict[str, Any], None]

Message content (must be string for SMS)

required
role str | None

Optional message role (not used in SMS channel)

None

Raises:

Type Description
TypeError

If response is not a string

RuntimeError

If the session or participant ids are missing

initiate_outbound_conversation async

initiate_outbound_conversation(
    options: InitiateMessagingConversationOptions,
) -> InitiateConversationResult

Initiate an outbound SMS conversation.

Creates a conversation via Conversation Orchestrator with inline participants, then sends the initial message via the Actions API. If an active conversation with the same addresses already exists (group-by dedup), CO returns 409 and the existing conversation is reused.

SMSChannelConfig

Bases: MessagingChannelConfig

Configuration for SMS channel.

Inherits dedup_capacity and memory_mode from MessagingChannelConfig.

VoiceChannel

VoiceChannel(
    tac: TAC,
    config: VoiceChannelConfig
    | dict[str, Any]
    | None = None,
)

Bases: BaseChannel

Voice Channel for handling voice-based conversations via WebSocket.

Key features: - TwiML generation for incoming calls (see twiml module) - WebSocket connection management for real-time voice streaming - Conversation lifecycle management (inherited from BaseChannel) - Outbound call initiation

This channel is framework-agnostic and accepts any WebSocket implementation satisfying WebSocketProtocol. For a batteries-included FastAPI server, use tac.server.TACFastAPIServer.

Initialize Voice channel for websocket protocol handling.

Parameters:

Name Type Description Default
tac TAC

TAC instance for memory/context operations

required
config VoiceChannelConfig | dict[str, Any] | None

Voice channel configuration (VoiceChannelConfig or dict). If None, uses default configuration.

None

Examples:

>>> channel = VoiceChannel(tac, config={"memory_mode": "always"})
>>> channel = VoiceChannel(tac, config=VoiceChannelConfig(session_manager=sm))
>>> channel = VoiceChannel(tac)  # Use defaults

on_inbound_call_twiml

on_inbound_call_twiml(
    callback: InboundCallTwiMLHandler,
) -> None

Register a callback that produces per-call overrides for the TwiML inside <ConversationRelay> on inbound calls.

The callback receives a framework-neutral TwiMLRequest (parsed from the Twilio webhook form) and returns a TwiMLOptions. Fields the callback explicitly sets override default_twiml_options and TAC defaults; unset fields fall through.

Example
async def by_country(req: TwiMLRequest) -> TwiMLOptions:
    if req.caller_country == "MX":
        return TwiMLOptions(language="es-MX", welcome_greeting="¡Hola!")
    return TwiMLOptions()


voice_channel.on_inbound_call_twiml(by_country)

Outbound calls don't use this — pass per-call TwiML via InitiateVoiceConversationOptions.twiml_options directly.

handle_incoming_call async

handle_incoming_call(
    twiml_request: TwiMLRequest | None = None,
    *,
    host_twiml_options: TwiMLOptions | None = None,
) -> str

Generate TwiML response for incoming voice calls.

ConversationRelay automatically handles conversation creation and participant management via the conversation_configuration parameter.

The WebSocket URL and default session-cleanup action URL are derived from TACConfig.voice_public_domain + TACConfig.voice_websocket_path / voice_action_path.

TwiML fields are merged per-field, highest precedence first: 1. Output of the customizer registered via VoiceChannel.on_inbound_call_twiml(...) if configured and twiml_request is given. (Application-owned.) 2. VoiceChannelConfig.default_twiml_options — per-channel defaults. 3. host_twiml_options — per-call transport facts supplied by the host (the code owning the route), e.g. a per-call websocket_url with an affinity token. 4. TAC defaults: a fixed default welcome_greeting, conversation_configuration from TACConfig, action_url resolved via Studio handoff (when studio_handoff_flow_sid is configured), else derived from TACConfig.voice_public_domain + voice_action_path, and the websocket_url derived from TACConfig.voice_public_domain + voice_websocket_path.

Fields not set at a layer fall through to lower layers. Lists (languages) and nested models (custom_parameters) replace wholesale when set at a higher-priority layer. websocket_url falls back to the TACConfig-derived URL if unset at every layer.

The two arguments are complementary, not alternatives — a custom host typically passes both on the same call: twiml_request carries the inbound call's data (so the application's customizer can run), and host_twiml_options carries the host's own per-call overrides.

Parameters:

Name Type Description Default
twiml_request TwiMLRequest | None

The incoming Twilio voice webhook, parsed into a framework-neutral form (From, To, CallSid, CallerCountry, …). Supplied by Twilio; forwarded to the on_inbound_call_twiml customizer so the application can produce per-call overrides.

None
host_twiml_options TwiMLOptions | None

Per-call TwiML overrides supplied by the host (the code owning the route — e.g. a custom server), for transport facts the SDK can't derive, such as a per-call websocket_url with an affinity token. Layered below default_twiml_options and the application customizer, so a developer's explicit settings still win.

None

Returns:

Type Description
str

TwiML XML string for call connection.

handle_conversation_relay_callback async

handle_conversation_relay_callback(
    payload_dict: dict[str, str],
) -> None

Handle ConversationRelay callback webhook from Twilio.

In relay-only mode, this is a secondary mechanism for cleaning up conversation state when a call ends (the primary mechanism is websocket disconnect). In orchestrated mode, conversation lifecycle is managed by CO webhooks, so this is a no-op.

Parameters:

Name Type Description Default
payload_dict dict[str, str]

Raw form data dict from the webhook request.

required

handle_websocket async

handle_websocket(websocket: WebSocketProtocol) -> None

Handle voice streaming WebSocket connection lifecycle.

This method manages the entire websocket connection: - Accepts the connection - Processes incoming messages - Tracks and cancels in-flight tasks (if session_manager provided) - Cleans up on disconnect

Parameters:

Name Type Description Default
websocket WebSocketProtocol

Any WebSocket implementation satisfying WebSocketProtocol

required

initiate_outbound_conversation async

initiate_outbound_conversation(
    options: InitiateVoiceConversationOptions,
) -> InitiateVoiceConversationResult

Initiate an outbound voice conversation.

Places an outbound call with inline TwiML that connects to ConversationRelay. The conversationConfiguration attribute tells CO to create and manage the conversation during passive hydration. The session is initialized lazily on the first prompt when the conversation is discovered by callSid.

TwiML fields are merged per-field, highest precedence first: 1. options.twiml_options — per-call overrides 2. VoiceChannelConfig.default_twiml_options — channel-wide defaults 3. TAC defaults: welcome greeting, conversation_configuration from TACConfig, and action_url from Studio handoff (if configured), else derived from TACConfig.voice_public_domain + voice_action_path.

Fields not set at a layer fall through to lower layers. Lists (languages) and nested models (custom_parameters) replace wholesale when set at a higher-priority layer.

The WebSocket URL is derived from TACConfig.voice_public_domain + TACConfig.voice_websocket_path, unless overridden per-call via options.websocket_url.

process_webhook async

process_webhook(
    webhook_data: dict[str, Any],
    idempotency_token: str | None = None,
) -> None

Process conversation webhooks for cleanup and cache invalidation.

Voice channel processes CONVERSATION_UPDATED events: - CLOSED status: Clean up local session state - INACTIVE status: Invalidate cached memory (memory will be updated by Conversation Orchestrator)

Note: Conversation tracking uses instance-local memory. In multi-instance deployments, webhooks may route to a different instance, preventing cleanup. See CLAUDE.md for horizontal scaling considerations.

Parameters:

Name Type Description Default
webhook_data dict[str, Any]

Raw webhook event data from Twilio

required
idempotency_token str | None

Optional Twilio idempotency token from request headers

None

send_response async

send_response(
    conversation_id: str,
    response: str
    | AsyncGenerator[str | dict[str, Any], None],
    role: str | None = None,
) -> None

Send voice response through the websocket connection for this conversation.

Supports both simple string responses and streaming async generators.

Parameters:

Name Type Description Default
conversation_id str

Conversation ID

required
response str | AsyncGenerator[str | dict[str, Any], None]

Response text (string) or async generator for streaming

required
role str | None

Optional message role (not used in this implementation, but kept for API consistency with BaseChannel interface)

None

get_websocket

get_websocket(
    conversation_id: str,
) -> WebSocketProtocol | None

Get the WebSocket connection for a specific conversation.

Parameters:

Name Type Description Default
conversation_id str

Conversation ID

required

Returns:

Type Description
WebSocketProtocol | None

WebSocket connection if exists, None otherwise

VoiceChannelConfig

Bases: BaseModel

Configuration for Voice channel.

TwiML configuration layers (highest precedence first):

Inbound calls (handle_incoming_call): 1. Output of the customizer registered via VoiceChannel.on_inbound_call_twiml(...) [optional] 2. default_twiml_options [optional] 3. handle_incoming_call(host_twiml_options=...) [optional] 4. TAC defaults

Outbound calls (initiate_outbound_conversation): 1. InitiateVoiceConversationOptions.twiml_options [optional] 2. default_twiml_options [optional] 3. TAC defaults

All layers merge per-field via Pydantic's model_fields_set — only fields a layer explicitly sets override lower layers. Lists (languages) and nested models (custom_parameters) replace wholesale when set.

Attributes:

Name Type Description
session_manager SessionManager | None

SessionManager for tracking and canceling in-flight tasks. Defaults to ThreadSafeSessionManager for automatic task cancellation on interrupts and new prompts. Set to None only for debugging/testing.

memory_mode MemoryMode

Memory retrieval mode. Default is "never". - "always": Retrieve memory for every message with the query string - "once": Retrieve memory once at conversation start with empty query and cache it. Cache is invalidated when conversation becomes INACTIVE. - "never": Skip memory retrieval

default_twiml_options TwiMLOptions | None

Static TwiMLOptions applied to every call (inbound and outbound). Controls the TwiML inside <ConversationRelay> — voice, language, transcription provider, welcome_greeting, <Language> children, etc. Use this when the same ConversationRelay configuration is correct for every call.

Per-call inbound customization is registered via VoiceChannel.on_inbound_call_twiml(...) (not on this config).

WhatsAppChannel

WhatsAppChannel(
    tac: TAC,
    config: WhatsAppChannelConfig
    | dict[str, Any]
    | None = None,
)

Bases: MessagingChannel

WhatsApp Channel for handling WhatsApp-based conversations.

Inherits shared messaging channel webhook processing from MessagingChannel and provides WhatsApp-specific message sending and filtering.

WhatsApp uses WhatsApp sender phone numbers configured in TACConfig (via TWILIO_WHATSAPP_NUMBER). Address format: whatsapp:+1234567890

is_default_agent_address

is_default_agent_address(author_address: str) -> bool

Check if the author address matches the configured WhatsApp number.

get_agent_address

get_agent_address(
    conversation_id: str,
) -> ParticipantAddress

Get the agent's participant address for this conversation.

send_response async

send_response(
    conversation_id: str,
    response: str
    | AsyncGenerator[str | dict[str, Any], None],
    role: str | None = None,
) -> None

Send WhatsApp response using the Conversation Orchestrator Send API.

Reads the agent and customer participant ids stashed on the session by inbound reconciliation or outbound initiation. Missing ids are a misuse — send_response is only expected to be called after an inbound webhook (COMMUNICATION_CREATED → reconcile) or after initiate_outbound_conversation, both of which populate the session.

Parameters:

Name Type Description Default
conversation_id str

Conversation ID to send response to

required
response str | AsyncGenerator[str | dict[str, Any], None]

Message content (must be string for WhatsApp)

required
role str | None

Optional message role (not used in WhatsApp channel)

None

Raises:

Type Description
TypeError

If response is not a string

RuntimeError

If the session or participant ids are missing

initiate_outbound_conversation async

initiate_outbound_conversation(
    options: InitiateMessagingConversationOptions,
) -> InitiateConversationResult

Initiate an outbound WhatsApp conversation.

Creates a conversation via Conversation Orchestrator with inline participants, then sends the initial message via the Actions API. Uses the WhatsApp number from TACConfig as the from address. If an active conversation with the same addresses already exists (group-by dedup), CO returns 409 and the existing conversation is reused.

Parameters:

Name Type Description Default
options InitiateMessagingConversationOptions

Conversation initiation options (to address and message)

required

Returns:

Type Description
InitiateConversationResult

InitiateConversationResult with conversation_id and session

Raises:

Type Description
RuntimeError

If whatsapp_number is not configured

WhatsAppChannelConfig

Bases: MessagingChannelConfig

Configuration for WhatsApp channel.

Inherits dedup_capacity and memory_mode from MessagingChannelConfig.