The CallManager class orchestrates the complete lifecycle of voice calls, from initiation through media streaming to call termination.
Overview
CallManager handles:
- Call initiation via Twilio REST API
- Incoming and outbound call management
- Media stream handling and audio bridging
- Real-time connection to Gemini Live API or OpenAI Realtime API
- Gateway communication for external integrations
- Session state tracking and cleanup
Constructor
from agenticai.core import CallManager
from agenticai.core.config import Config
call_manager = CallManager(config)
Application configuration object containing Twilio, Gemini/OpenAI, and Gateway settings.
Properties
active_sessions
@property
def active_sessions(self) -> dict[str, CallSession]
Dictionary mapping call IDs to their corresponding CallSession objects.
Methods
start
async def start(self) -> None
await call_manager.start()
stop
async def stop(self) -> None
await call_manager.stop()
initiate_call
async def initiate_call(
self,
to_number: str,
prompt: str,
webhook_base_url: str,
metadata: dict | None = None,
) -> str
Phone number to call in E.164 format (e.g., +1234567890).
System instruction/prompt for the AI agent during the call.
Base URL for Twilio webhooks (e.g., https://example.com).
Optional metadata to attach to the call session.
Unique identifier for the initiated call.
from agenticai.core import CallManager
from agenticai.core.config import get_config
import asyncio
async def make_call():
config = get_config("config.yaml")
call_manager = CallManager(config)
await call_manager.start()
try:
call_id = await call_manager.initiate_call(
to_number="+1234567890",
prompt="You are a helpful AI assistant.",
webhook_base_url="https://example.com",
metadata={"purpose": "customer_support"}
)
print(f"Call initiated: {call_id}")
# Wait for call to complete
while call_id in call_manager.active_sessions:
await asyncio.sleep(1)
finally:
await call_manager.stop()
asyncio.run(make_call())
register_incoming_call
async def register_incoming_call(
self,
call_sid: str,
from_number: str,
to_number: str,
) -> str
Twilio call SID for the incoming call.
Your Twilio number that was called.
Unique call ID for the new session.
call_id = await call_manager.register_incoming_call(
call_sid="CA1234567890abcdef",
from_number="+1234567890",
to_number="+1987654321"
)
handle_call_status
async def handle_call_status(self, call_sid: str, status: str) -> None
New call status (e.g., completed, failed, busy, no-answer, canceled).
await call_manager.handle_call_status(
call_sid="CA1234567890abcdef",
status="completed"
)
async def handle_media_stream(self, twilio_handler: TwilioMediaStreamHandler) -> None
twilio_handler
TwilioMediaStreamHandler
required
Twilio WebSocket handler for the media stream.
from agenticai.twilio.websocket import TwilioMediaStreamHandler
twilio_handler = TwilioMediaStreamHandler(websocket)
await call_manager.handle_media_stream(twilio_handler)
end_call
async def end_call(self, call_id: str) -> None
await call_manager.end_call("550e8400-e29b-41d4-a716-446655440000")
get_pending_call_info
def get_pending_call_info(self, call_sid: str) -> dict | None
Dictionary containing call_id, prompt, and metadata, or None if not found.
CallSession Data Class
The CallSession class represents an active call session:
@dataclass
class CallSession:
call_id: str # Unique call identifier
call_sid: str # Twilio call SID
to_number: str # Phone number being called
prompt: str # AI system prompt
metadata: dict # Additional metadata
start_time: datetime # Call start timestamp
bridge: AudioBridge | None # Audio bridge instance
status: str # Call status (initiating, ringing, in-progress, completed, etc.)