test_outbound_call.py

#!/usr/bin/env python3
"""
Script to test outbound calling functionality.
This dispatches the agent to place an outbound call.
"""

import asyncio
import os
import sys
import json
import random
# from dotenv import load_dotenv
from livekit import api
from livekit.protocol.sip import CreateSIPParticipantRequest, SIPParticipantInfo


# Load environment variables
# load_dotenv(".env.local")


async def make_outbound_call(phone_number: str, sip_trunk_id: str = None):
    """
    Dispatch the agent to make an outbound call.

    Args:
        phone_number: The phone number to call (e.g., "+15105550123")
        sip_trunk_id: The SIP trunk ID to use (optional, can be set in metadata)
    """

    # Get LiveKit credentials from environment
    livekit_url = "https://test-v0gstj3o.livekit.cloud"
    livekit_api_key = "APIsxDLvM5EywoZ"
    livekit_api_secret = "Ekr4w73mdf1eWNjxjAVzwwfMX6jyt1nxyVucrkclUpQB"


    # Create LiveKit API client
    lkapi = api.LiveKitAPI(
        url=livekit_url,
        api_key=livekit_api_key,
        api_secret=livekit_api_secret
    )

    # Generate a unique room name
    room_name = f"outbound-{''.join(str(random.randint(0, 9)) for _ in range(10))}"

    # Prepare metadata with phone number and SIP trunk ID
    metadata = {
        "phone_number": phone_number
    }

    if sip_trunk_id:
        metadata["sip_trunk_id"] = sip_trunk_id


    try:
        print(f"📞 Dispatching agent to call {phone_number}...")
        print(f"   Room: {room_name}")

        # Create dispatch to trigger the agent
        # response = await lkapi.agent_dispatch.create_dispatch(
        #     api.CreateAgentDispatchRequest(
        #         agent_name="my-telephony-agent",
        #         room=room_name,
        #         metadata=json.dumps(metadata)
        # ))

        #sends agent to room, but does not place the call
        response = await lkapi.agent_dispatch.create_dispatch(
            api.CreateAgentDispatchRequest(
                agent_name="my-telephony-agent",
                room=room_name,
        ))

        #places the call
        request = CreateSIPParticipantRequest(
                    sip_trunk_id = sip_trunk_id,
                    sip_call_to = phone_number,
                    room_name = room_name,
                    participant_identity = "sip-test",
                    participant_name = "Test Caller",
                    krisp_enabled = True,
                    wait_until_answered = True
                )
    
        try:
            participant = await lkapi.sip.create_sip_participant(request)
            print(f"Successfully created {participant}")
        except Exception as e:
            print(f"Error creating SIP participant: {e}")
        finally:
            await lkapi.aclose()



        print(f"✅ Agent dispatched successfully!")
        print(f"   Room: {room_name}")
        print(f"   Dispatch ID: {response.dispatch_id if hasattr(response, 'dispatch_id') else 'N/A'}")
        print(f"\n   The agent is now placing the call to {phone_number}")
        print(f"   Monitor your agent logs to see the call progress")

    except Exception as e:
        print(f"❌ Failed to dispatch agent: {e}")
        sys.exit(1)


async def main():
    """Main entry point for the script."""

    print("🚀 LiveKit Outbound Call Test")
    print("="*50)

    phone_number = "+13179854301"
    sip_trunk_id = "ST_XAbYyQgcKzMm"

    # Make the call
    await make_outbound_call(phone_number, sip_trunk_id)


if __name__ == "__main__":
    asyncio.run(main())