Manager API Connection and Actions

class Manager(BaseManager, CoreActions, ZapataActions):
    """
    Full-featured Manager API client with protocol implementation, 
    core actions, Zapata actions, and event handling.
    """
    def __init__(address: tuple, username: str, secret: str, 
                 listen_events: bool = True, timeout: float = None):
        """
        Connect to Asterisk Manager API.
        
        Args:
            address: (host, port) tuple for Manager API connection
            username: Manager API username  
            secret: Manager API password
            listen_events: Whether to receive real-time events
            timeout: Socket timeout in seconds
        """

class CoreManager(BaseManager, CoreActions, ZapataActions):
    """
    Manager API client with actions but without event handlers.
    Use when you don't need event processing.
    """

class BaseManager:
    """
    Base protocol implementation for Manager API communication.
    """
    def __init__(address: tuple, username: str, secret: str, 
                 listen_events: bool = True, timeout: float = None): ...
    
    def get_channel(channel_id: str) -> BaseChannel:
        """Get a channel object for the given channel ID."""
    
    def read_response(action_id: str) -> dict:
        """Read response for a specific action ID."""

def Login(username: str, secret: str) -> dict:
    """Authenticate with the Manager API."""

def Logoff() -> dict:
    """Disconnect from the Manager API."""

def Command(command: str) -> list:
    """Execute Asterisk console command and return output lines."""

def Ping() -> dict:
    """Send keepalive ping to server."""

def ListCommands() -> dict:
    """Get list of available Manager API commands."""

def CoreShowChannels() -> list:
    """Get information about all active channels."""

def Status(channel: str = None) -> dict:
    """Get status of specific channel or all channels."""

def Hangup(channel: str) -> dict:
    """Hang up the specified channel."""

def AbsoluteTimeout(channel: str, timeout: int) -> dict:
    """Set absolute timeout for channel."""

def Originate(channel: str, context: str, exten: str, priority: int,
              timeout: int = None, callerid: str = None,
              variable: dict = None, **kwargs) -> dict:
    """Originate a new call."""

def Getvar(channel: str, variable: str) -> dict:
    """Get value of channel variable."""

def Setvar(channel: str, variable: str, value: str) -> dict:
    """Set channel variable to value."""

def Monitor(channel: str, filename: str = None, format: str = 'wav',
            mix: bool = False) -> dict:
    """Start monitoring/recording channel."""

def StopMonitor(channel: str) -> dict:
    """Stop monitoring channel."""

def ChangeMonitor(channel: str, pathname: str) -> dict:
    """Change monitor filename for channel."""

def QueueStatus(queue: str = None) -> dict:
    """Get status of call queues."""

def QueueAdd(queue: str, interface: str, penalty: int = 0,
             paused: bool = False, membername: str = None) -> dict:
    """Add member to call queue."""

def QueueRemove(queue: str, interface: str) -> dict:
    """Remove member from call queue."""

def ConfbridgeListRooms() -> list:
    """
    List active conference rooms.
    Returns list of dicts with room_name, users_count, marked_users, locked.
    """

def ConfbridgeList(room: str) -> list:
    """
    List participants in conference room.
    Returns list of dicts with channel, user_profile, bridge_profile, 
    menu, caller_id, muted.
    """

def ConfbridgeKick(room: str, channel: str) -> bool:
    """Kick channel from conference room."""

def ConfbridgeMute(room: str, channel: str) -> bool:
    """Mute channel in conference room."""

def ConfbridgeUnmute(room: str, channel: str) -> bool:
    """Unmute channel in conference room."""

def ConfbridgeStartRecord(room: str, rFile: str = None) -> bool:
    """Start recording conference room."""

def ConfbridgeStopRecord(room: str) -> bool:
    """Stop recording conference room."""

def ConfbridgeisRecording(room: str) -> bool:
    """Check if conference room is being recorded."""

def MeetMe() -> list:
    """
    List active MeetMe conferences.
    Returns list of dicts with conference details.
    """

def MeetMeList(confnum: str) -> list:
    """
    List participants in MeetMe conference.
    
    Args:
        confnum: Conference number
        
    Returns:
        List of dicts with participant details
    """

def SipShowPeer(peer: str) -> dict:
    """Show detailed information about SIP peer."""

def SipShowRegistry() -> dict:
    """Show SIP registration status."""

def SipPeers() -> dict:
    """List all SIP peers with their status."""

def Reload(module: str = None) -> dict:
    """Reload Asterisk configuration or specific module."""

def ExtensionState(exten: str, context: str) -> dict:
    """Get state of extension."""

def ParkedCalls() -> dict:
    """Get list of parked calls."""

def SetCDRUserField(channel: str, userfield: str) -> dict:
    """Set CDR user field for channel."""

def VoicemailUsersList(context: str = None) -> dict:
    """List voicemail users."""

def DBGet(family: str, key: str) -> dict:
    """Get value from Asterisk database."""

def DBPut(family: str, key: str, value: str) -> dict:
    """Set value in Asterisk database."""

def Events(categories: str) -> dict:
    """Filter events to specific categories."""

def ExtensionStates() -> dict:
    """Get extension states for all extensions."""

def MailboxCount(mailbox: str) -> tuple:
    """Get mailbox message count as (new, old) tuple."""

def MailboxStatus(mailbox: str) -> int:
    """Get mailbox status as integer."""

def Bridge(channel1: str, channel2: str, tone: bool = False) -> dict:
    """Bridge two channels together."""

def PlayDTMF(channel: str, digit: str) -> dict:
    """Play DTMF digit on channel."""

def Originate2(channel: str, parameters: dict) -> dict:
    """Originate call using parameter dictionary."""

def Queues() -> dict:
    """Alias for QueueStatus() - get all queue statuses."""

class ZapataActions:
    def ZapShowChannels() -> dict:
        """Show Zapata/DAHDI channel information."""
    
    def ZapHangup(channel: int) -> dict:
        """Hang up Zapata channel by number."""
    
    def ZapTransfer(channel: int) -> dict:
        """Transfer Zapata channel."""
    
    def ZapDialOffhook(channel: int, number: str) -> dict:
        """Dial number on off-hook Zapata channel."""
    
    def ZapDNDon(channel: int) -> dict:
        """Enable Do Not Disturb on Zapata channel."""
    
    def ZapDNDoff(channel: int) -> dict:
        """Disable Do Not Disturb on Zapata channel."""

from Asterisk.Manager import Manager

# Connect to local Asterisk instance
manager = Manager(('127.0.0.1', 5038), 'admin', 'secret123')

try:
    # Execute commands
    channels = manager.CoreShowChannels()
    print(f"Active channels: {len(channels)}")
    
    # Originate a call
    result = manager.Originate(
        channel='SIP/1001',
        context='default', 
        exten='1002',
        priority=1,
        timeout=30,
        callerid='Test Call <1001>'
    )
    
finally:
    manager.Logoff()

# Check queue status
queues = manager.QueueStatus()
for queue_name, queue_info in queues.items():
    print(f"Queue {queue_name}: {queue_info.get('calls')} calls waiting")

# Add member to queue
manager.QueueAdd('support', 'SIP/1001', penalty=1, membername='Agent 1')