Core¶

tac.core ¶

Core TAC functionality.

TACConfig ¶

Bases: BaseModel

Configuration model for Twilio Agent Connect settings.

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 ¶

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 ¶

from_env `classmethod` ¶

retrieve_memory `async` ¶

process_cintel_event `async` ¶

trigger_error `async` ¶