Core¶

tac.core ¶

Core TAC functionality.

TACConfig `pydantic-model` ¶

Bases: BaseModel

Configuration model for Twilio Agent Connect settings.

Config:

use_enum_values: True
json_schema_extra: {'example': {'conversation_configuration_id': 'conv_configuration_xxxxxxxxxxxxxxxxxx', 'account_sid': 'ACxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx', 'auth_token': 'your_auth_token_here', 'api_key': 'SKxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx', 'api_secret': 'your_api_secret_here', 'phone_number': 'your_phone_number_here', 'memory_config': {'trait_groups': ['Contact', 'Preferences']}, 'conversation_intelligence_config': {'configuration_id': 'GAxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx', 'summary_operator_sid': 'LYyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy'}}}

Fields:

conversation_configuration_id (str | None)
memory_config (TwilioMemoryConfig)
account_sid (str)
auth_token (str)
api_key (str)
api_secret (str)
region (str | None)
phone_number (str)
rcs_sender_id (str | None)
whatsapp_number (str | None)
knowledge_base_id (str | None)
log_level (str)
studio_handoff_flow_sid (str | None)
voice_public_domain (str | None)
voice_websocket_path (str)
voice_action_path (str)
voice_call_event_path (str)
conversation_intelligence_config (ConversationIntelligenceConfig | None)

Validators:

_normalize_and_validate_region → region
_normalize_voice_public_domain → voice_public_domain

conversation_configuration_id `pydantic-field` ¶

conversation_configuration_id: str | None = None

Twilio Conversation Configuration ID. When omitted, TAC runs in ConversationRelay-only mode: only the Voice channel is usable, messaging channels cannot be constructed, and TAC.retrieve_memory() returns an empty TACMemoryResponse.

memory_config `pydantic-field` ¶

memory_config: TwilioMemoryConfig

Twilio Memory configuration for controlling retrieval limits, relevance threshold, and trait groups. Memory client is always initialized automatically from Conversation Orchestrator configuration.

account_sid `pydantic-field` ¶

account_sid: str

Twilio Account SID

auth_token `pydantic-field` ¶

auth_token: str

Twilio Auth Token

api_key `pydantic-field` ¶

api_key: str

Twilio API Key SID (starts with SK)

api_secret `pydantic-field` ¶

api_secret: str

Twilio API Key Secret

region `pydantic-field` ¶

region: str | None = None

Optional Twilio region (e.g., 'au1', 'ie1'). When set, API base URLs become https://product..twilio.com

phone_number `pydantic-field` ¶

phone_number: str

Twilio Phone Number for Voice (inbound) and SMS (send/receive).

rcs_sender_id `pydantic-field` ¶

rcs_sender_id: str | None = None

Optional Twilio RCS Sender ID

whatsapp_number `pydantic-field` ¶

whatsapp_number: str | None = None

Optional Twilio WhatsApp-enabled phone number (format: whatsapp:+1234567890)

knowledge_base_id `pydantic-field` ¶

knowledge_base_id: str | None = None

Optional Knowledge Base ID for knowledge search functionality

log_level `pydantic-field` ¶

log_level: str = 'INFO'

Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)

studio_handoff_flow_sid `pydantic-field` ¶

studio_handoff_flow_sid: str | None = None

Twilio Studio Flow SID (FWxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx) for handoff. TAC constructs both the digital-handoff Studio Executions URL (studio.twilio.com/v2/Flows/{SID}/Executions) and the voice webhook URL (webhooks.twilio.com/v1/Accounts/{AccountSid}/Flows/{SID}?Trigger=incomingCall) from this SID.

voice_public_domain `pydantic-field` ¶

voice_public_domain: str | None = None

Public domain where voice routes are reachable (e.g. 'example.ngrok.app'). Used by VoiceChannel to construct the public WebSocket URL and ConversationRelay action URL. Required when using the Voice channel. Schemes (https://, wss://) and trailing slashes are stripped automatically.

voice_websocket_path `pydantic-field` ¶

voice_websocket_path: str = '/ws'

Path the voice WebSocket is served at. Combined with voice_public_domain to build the public WebSocket URL the voice channel hands to Twilio in TwiML; TACFastAPIServer also registers its WebSocket route at this path. Override only if you mount the route at a non-default path.

voice_action_path `pydantic-field` ¶

voice_action_path: str = '/conversation-relay-callback'

Path the ConversationRelay action callback is served at. Same role as voice_websocket_path but for the cleanup callback.

voice_call_event_path `pydantic-field` ¶

voice_call_event_path: str = '/twilio/call-events'

Base path for the call-event callbacks (status, async AMD, recording). TACFastAPIServer registers one route per callback under it — /status, /amd, /recording — so the route identifies the event. Same role as voice_action_path.

conversation_intelligence_config `pydantic-field` ¶

conversation_intelligence_config: (
    ConversationIntelligenceConfig | None
) = None

Optional Conversation Intelligence configuration for filtering webhook events. When provided to OperatorResultProcessor, only matching events are processed.

call_event_path ¶

call_event_path(kind: CallEventKind) -> str

Path a call-event callback is served at.

Single source of truth: the channel builds callback URLs from this and TACFastAPIServer registers routes at it, so the two can't drift.

call_event_url ¶

call_event_url(kind: CallEventKind) -> str | None

Public URL for a call-event callback, or None without a public domain.

from_env `classmethod` ¶

from_env() -> TACConfig

Create TACConfig from environment variables.

Required:

TWILIO_ACCOUNT_SID: Twilio Account SID
TWILIO_AUTH_TOKEN: Twilio Auth Token for API authentication
TWILIO_API_KEY: Twilio API Key SID (starts with SK)
TWILIO_API_SECRET: Twilio API Secret for API Key authentication
TWILIO_PHONE_NUMBER: Phone number for voice and SMS channels

Required for Conversation Orchestrator / Memory / Knowledge:

TWILIO_CONVERSATION_CONFIGURATION_ID: Conversation Orchestrator configuration ID (when omitted, TAC runs in ConversationRelay-only mode)

Optional:

TWILIO_RCS_SENDER_ID: RCS Sender ID for RCS channel
TWILIO_WHATSAPP_NUMBER: WhatsApp-enabled phone number (format: whatsapp:+1234567890)
TWILIO_KNOWLEDGE_BASE_ID: Knowledge Base ID for RAG search functionality
TWILIO_LOG_LEVEL: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL). Default: INFO
TWILIO_REGION: Twilio region for data residency (e.g., 'au1', 'ie1')
TWILIO_STUDIO_HANDOFF_FLOW_SID: Studio Flow SID (FWxxx...) for handoff tool
TWILIO_VOICE_PUBLIC_DOMAIN: Public domain for voice routes (required for voice)
TWILIO_VOICE_WEBSOCKET_PATH: Path for voice WebSocket (default: /ws)
TWILIO_VOICE_ACTION_PATH: Path for ConversationRelay action callback (default: /conversation-relay-callback)

Memory Configuration:

TWILIO_MEMORY_PROFILE_TRAIT_GROUPS: Trait groups to include (comma-separated, e.g., "Contact,Preferences")
TWILIO_MEMORY_OBSERVATIONS_LIMIT: Max observations in memory retrieval. Default: 20
TWILIO_MEMORY_SUMMARIES_LIMIT: Max summaries in memory retrieval. Default: 5
TWILIO_MEMORY_COMMUNICATIONS_LIMIT: Max communications in memory retrieval. Default: 0
TWILIO_MEMORY_RELEVANCE_THRESHOLD: Min relevance score (0.0-1.0). Default: 0.0

Conversation Intelligence:

CONVERSATION_INTELLIGENCE_CONFIGURATION_ID: CI Service configuration ID for webhook filtering
CONVERSATION_INTELLIGENCE_SUMMARY_OPERATOR_SID: Operator SID for summary extraction

TAC ¶

TAC(config: TACConfig | dict[str, Any])

Main Twilio Agent Connect class for processing webhook events with configuration.

This class accepts configuration and provides methods to process webhook events.

Initialize TAC instance with configuration.

Parameters:

Name	Type	Description	Default
`config`	`TACConfig \| dict[str, Any]`	TACConfig instance or dictionary with configuration parameters.	required

retrieve_memory `async` ¶

retrieve_memory(
    conversation_context: ConversationSession,
    query: str | None = None,
    conversation_id: str | None = None,
) -> TACMemoryResponse

Retrieve memories from Memory Store with fallback to Conversation Orchestrator.

Three-tier resolution: 1. Memory API (when conversation_memory_client is configured). 2. Conversation Orchestrator list_communications fallback (when CO is configured). 3. Empty TACMemoryResponse (relay-only mode).

Parameters:

Name	Type	Description	Default
`conversation_context`	`ConversationSession`	Session containing conversation and profile information.	required
`query`	`str \| None`	Optional search query to filter memories.	`None`
`conversation_id`	`str \| None`	Passed through to the /Recall request as-is; the caller decides whether (and which) conversation_id to send. Passing one without a query makes Twilio Memory infer a query from that conversation's history — an expensive server-side step — so leave this unset for fetches with no per-turn topic to justify that cost (e.g. "once" mode's cache-priming fetch).	`None`

Returns:

Type	Description
`TACMemoryResponse`	Memory response containing conversation history and profile data.

process_cintel_event `async` ¶

process_cintel_event(
    payload: dict[str, Any],
) -> OperatorProcessingResult

Process Conversation Intelligence webhook and create observations/summaries in Memory.

Parameters:

Name	Type	Description	Default
`payload`	`dict[str, Any]`	Webhook payload from Conversation Intelligence service.	required

Returns:

Type	Description
`OperatorProcessingResult`	Processing result with created observations and summaries.

on_message_ready ¶

on_message_ready(
    callback: Callable[
        [
            str,
            ConversationSession,
            TACMemoryResponse | None,
        ],
        str | None,
    ]
    | Callable[
        [
            str,
            ConversationSession,
            TACMemoryResponse | None,
        ],
        Awaitable[str | None],
    ],
) -> None

Register callback invoked when a message is ready.

Callback can return a string (TAC auto-sends to channel) or None (manual handling).

Example

async def handle_message(
    message: str, context: ConversationSession, memory: TACMemoryResponse | None
) -> str:
    response = await openai_client.responses.create(...)
    return response.output_text  # TAC routes to appropriate channel


tac.on_message_ready(handle_message)

Parameters:

Name	Type	Description	Default
`callback`	`Callable[[str, ConversationSession, TACMemoryResponse \| None], str \| None] \| Callable[[str, ConversationSession, TACMemoryResponse \| None], Awaitable[str \| None]]`	Function with (message, context, memory). Returns str or None.	required

on_interrupt ¶

on_interrupt(
    callback: Callable[[ConversationSession, Any], None]
    | Callable[[ConversationSession, Any], Awaitable[None]],
) -> None

Register callback invoked on user interrupt.

Example

def handle_interrupt(context: ConversationSession, interrupt_data: Any):
    # Handle user interrupt...
    pass


tac.on_interrupt(handle_interrupt)

Parameters:

Name	Type	Description	Default
`callback`	`Callable[[ConversationSession, Any], None] \| Callable[[ConversationSession, Any], Awaitable[None]]`	Function to call with (context, interrupt_data). Supports sync and async.	required

on_conversation_ended ¶

on_conversation_ended(
    callback: Callable[[ConversationSession], None]
    | Callable[[ConversationSession], Awaitable[None]],
) -> None

Register callback invoked when conversation ends.

Example

def handle_conversation_ended(context: ConversationSession):
    # Clean up conversation...
    pass


tac.on_conversation_ended(handle_conversation_ended)

Parameters:

Name	Type	Description	Default
`callback`	`Callable[[ConversationSession], None] \| Callable[[ConversationSession], Awaitable[None]]`	Function to call with conversation context. Supports sync and async.	required

on_error ¶

on_error(
    callback: Callable[[Exception, dict[str, Any]], None]
    | Callable[
        [Exception, dict[str, Any]], Awaitable[None]
    ],
) -> None

Register callback invoked when TAC encounters a recoverable error.

Use this to surface failures that TAC handles internally but that would otherwise be invisible to your application — for example, an inbound message that is dropped because participant reconciliation failed (Conversation Orchestrator unreachable, a participant-promotion conflict, or no resolvable customer). Without a handler these are only logged; registering on_error lets you alert, retry, or reconcile.

The callback receives the exception and a context dict with details such as conversation_id, channel, and dropped_inbound (True when an inbound message was discarded). Keys present depend on the error site, so read them defensively.

Example

def handle_error(error: Exception, context: dict) -> None:
    if context.get("dropped_inbound"):
        alert(f"Dropped inbound on {context.get('conversation_id')}: {error}")


tac.on_error(handle_error)

Parameters:

Name	Type	Description	Default
`callback`	`Callable[[Exception, dict[str, Any]], None] \| Callable[[Exception, dict[str, Any]], Awaitable[None]]`	Function to call with (error, context). Supports sync and async.	required

trigger_error `async` ¶

trigger_error(
    error: Exception, context: dict[str, Any]
) -> None

Trigger the registered error callback.

No-op when no handler is registered, so callers can invoke it unconditionally. The handler is invoked defensively: an exception raised by the handler itself is logged and swallowed so error reporting never breaks the caller's flow.

Parameters:

Name	Type	Description	Default
`error`	`Exception`	The exception that occurred.	required
`context`	`dict[str, Any]`	Details about the error site (e.g. `conversation_id`, `channel`, `dropped_inbound`).	required

get_logger ¶

get_logger(name: str, **context: Any) -> ContextLogger

Get a context-aware logger instance for a specific module.

Parameters:

Name	Type	Description	Default
`name`	`str`	Logger name (typically name from the calling module)	required
`**context`	`Any`	Initial context to bind (e.g., conversation_id, channel)	`{}`

Returns:

Type	Description
`ContextLogger`	ContextLogger instance with bound context

Core¶

tac.core ¶

TACConfig pydantic-model ¶

conversation_configuration_id pydantic-field ¶

memory_config pydantic-field ¶

account_sid pydantic-field ¶

auth_token pydantic-field ¶

api_key pydantic-field ¶

api_secret pydantic-field ¶

region pydantic-field ¶

phone_number pydantic-field ¶

rcs_sender_id pydantic-field ¶

whatsapp_number pydantic-field ¶

knowledge_base_id pydantic-field ¶

log_level pydantic-field ¶

studio_handoff_flow_sid pydantic-field ¶

voice_public_domain pydantic-field ¶

voice_websocket_path pydantic-field ¶

voice_action_path pydantic-field ¶

voice_call_event_path pydantic-field ¶

conversation_intelligence_config pydantic-field ¶

call_event_path ¶

call_event_url ¶

from_env classmethod ¶

TAC ¶

retrieve_memory async ¶

process_cintel_event async ¶

on_message_ready ¶

on_interrupt ¶

on_conversation_ended ¶

on_error ¶

trigger_error async ¶

get_logger ¶

TACConfig `pydantic-model` ¶

conversation_configuration_id `pydantic-field` ¶

memory_config `pydantic-field` ¶

account_sid `pydantic-field` ¶

auth_token `pydantic-field` ¶

api_key `pydantic-field` ¶

api_secret `pydantic-field` ¶

region `pydantic-field` ¶

phone_number `pydantic-field` ¶

rcs_sender_id `pydantic-field` ¶

whatsapp_number `pydantic-field` ¶

knowledge_base_id `pydantic-field` ¶

log_level `pydantic-field` ¶

studio_handoff_flow_sid `pydantic-field` ¶

voice_public_domain `pydantic-field` ¶

voice_websocket_path `pydantic-field` ¶

voice_action_path `pydantic-field` ¶

voice_call_event_path `pydantic-field` ¶

conversation_intelligence_config `pydantic-field` ¶

from_env `classmethod` ¶

retrieve_memory `async` ¶

process_cintel_event `async` ¶

trigger_error `async` ¶