connexity 1.0.3

A Python client for Connexity API.

Connexity SDK for Pipecat

A Python SDK for tracking and analyzing voice AI call sessions with Pipecat. Provides frame observers for Twilio and Daily.co integrations to capture conversation data, latency metrics, and call analytics.

Installation

pip install connexity

Version

from connexity import __version__
print(__version__)

Usage

Twilio Observer

Use ConnexityTwilioObserver for Twilio-based telephony calls:

from pipecat.audio.vad.vad_analyzer import VADParams
from connexity.pipecat.twilio import ConnexityTwilioObserver
from connexity.utils.logging_config import LogLevel
from twilio.rest import Client

# Configure VAD parameters
vad_params = VADParams(
    confidence=0.5,
    min_volume=0.6,
    start_secs=0.2,
    stop_secs=0.8,
)

# Initialize Twilio client and start recording
twilio_client = Client(account_sid, auth_token)
twilio_client.calls(call_sid).recordings.create()

# Create and initialize the observer
observer = ConnexityTwilioObserver()
await observer.initialize(
    sid=call_sid,                         # Twilio Call SID
    agent_id="YOUR_AGENT_ID",             # Your Connexity agent ID
    api_key="YOUR_CONNEXITY_API_KEY",     # Your Connexity API key
    vad_params=vad_params,                # VAD configuration
    run_mode="production",                # "development" or "production"
    vad_analyzer="silero",                # VAD engine name
    twilio_client=twilio_client,          # Twilio REST client instance
    log_level=LogLevel.INFO,              # Optional: DEBUG, INFO, WARNING, ERROR, CRITICAL
    latency_threshold_ms=4000.0,          # Optional: latency alert threshold
)

# Register with your Pipecat pipeline
pipeline.register_observer(observer)

Daily.co Observer

Use ConnexityDailyObserver for Daily.co-based WebRTC calls:

from pipecat.audio.vad.vad_analyzer import VADParams
from connexity.pipecat.daily import ConnexityDailyObserver
from connexity.utils.logging_config import LogLevel

# Configure VAD parameters
vad_params = VADParams(
    confidence=0.5,
    min_volume=0.6,
    start_secs=0.2,
    stop_secs=0.8,
)

# Create and initialize the observer
observer = ConnexityDailyObserver()
await observer.initialize(
    sid=room_name,                        # Daily.co room name/ID
    agent_id="YOUR_AGENT_ID",             # Your Connexity agent ID
    api_key="YOUR_CONNEXITY_API_KEY",     # Your Connexity API key
    vad_params=vad_params,                # VAD configuration
    run_mode="production",                # "development" or "production"
    vad_analyzer="silero",                # VAD engine name
    daily_api_key="YOUR_DAILY_API_KEY",   # Daily.co API key (required)
    log_level=LogLevel.INFO,              # Optional: DEBUG, INFO, WARNING, ERROR, CRITICAL
    latency_threshold_ms=4000.0,          # Optional: latency alert threshold
)
# Note: Daily.co calls are always treated as "web" type with no phone numbers

# Register with your Pipecat pipeline
pipeline.register_observer(observer)

Tool Observer

Use @observe_tool decorator to monitor and track tool function executions:

from connexity.pipecat import observe_tool
from pipecat.services.llm_service import FunctionCallParams
from typing import Dict

@observe_tool(
    tool_name="end_call",              # Optional: explicit tool name
    include_result=True,                # Optional: capture return value (default: True)
    include_traceback=True,            # Optional: capture tracebacks on errors (default: True)
    enable_timeout=True,               # Optional: enforce execution timeout (default: True)
    timeout_seconds=10.0,              # Optional: timeout duration in seconds (default: 10.0)
)
async def end_call(params: FunctionCallParams) -> Dict[str, str]:
    # Your tool implementation
    return {"status": "ended", "sid": call_sid}

The decorator automatically tracks:

• Execution timing and duration
• Tool call IDs and arguments
• Success/failure status
• Error detection in results
• Timeout enforcement
• Callback invocation tracking

Logging Configuration

The SDK uses a centralized logging system that can be configured globally:

from connexity.utils.logging_config import set_sdk_log_level, LogLevel

# Set log level for all SDK components
set_sdk_log_level(LogLevel.DEBUG)

Available log levels: DEBUG, INFO, WARNING, ERROR, CRITICAL

Features

• Conversation Capture: Records user/assistant messages with timing
• Latency Tracking: Measures STT, LLM, TTS, and end-to-end latency
• Interruption Detection: Identifies unsuccessful user interruptions and interruption loops
• Tool Call Monitoring: Tracks function call lifecycle and issues
• Issue Reporting: Automatically reports latency peaks and errors
• System Prompt Extraction: Captures and analyzes system prompts
• Recording Integration: Retrieves call recordings from Twilio/Daily.co