Documentation Index
Fetch the complete documentation index at: https://mintlify.com/IstiqlalBhat/aiagent/llms.txt
Use this file to discover all available pages before exploring further.
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.)