Making calls using SIP

This guide walks you through the steps to create an AI voice agent that makes outgoing calls. You can use an agent in the following scenarios:

Call center callbacks: Automatically return customer calls.
Product check-ins: Follow up on purchased items.
Sales calls: Reach out to potential customers.
Appointment reminders: Notify clients of upcoming appointments.
Surveys: Collect feedback through automated calls.

Prerequisites

The following are required to complete the steps in this guide:

Phone number purchased from a SIP trunk provider like Twilio or Telnyx.
SIP provider trunk configured as an outbound trunk for use with LiveKit SIP.
LiveKit Cloud project or a self-hosted instance of LiveKit server
SIP server (only required if you're self hosting the LiveKit server).
LiveKit CLI installed (requires version 2.3.0 or later).

Outbound call flow

The suggested flow for outbound calls is as follows:

Create a dispatch for your agent.
Once your agent is connected, dial the user via CreateSIPParticipant.
The user answers the call, and starts speaking to the agent.

Step 1: Set up environment variables

tip

Set up the following environment variables to configure the LiveKit CLI to use your LiveKit Cloud or self-hosted LiveKit server instance:

export LIVEKIT_URL=<your LiveKit server URL>
export LIVEKIT_API_KEY=<your API Key>
export LIVEKIT_API_SECRET=<your API Secret>
export OPENAI_API_KEY=your-openai-api-key

Step 2: Create an outbound trunk

To make outgoing calls, the provider's outbound trunk needs to be registered with LiveKit. SIP trunking providers typically require authentication when accepting outbound SIP requests to ensure only authorized users are making calls with your number.

note

This setup only needs to be performed once.

If you already have an outbound trunk for the SIP provider phone number, you can skip to Step 3: Create an agent.

Create a file named outbound-trunk.json using your phone number, trunk domain name, and username and password. The following example assumes your provider phone number is +15105550100:

{
  "trunk": {
    "name": "My outbound trunk",
    "address": "<my-trunk>.pstn.twilio.com",
    "numbers": ["+15105550100"],
    "auth_username": "<username>",
    "auth_password": "<password>"
  }
}

Create the outbound trunk using the CLI:
```
lk sip outbound create outbound-trunk.json
```
The output of the command returns the trunk ID. Copy the <trunk-id> for step 4:
```
SIPTrunkID: <trunk-id>
```

Step 3: Create an agent

Create an agent that makes outbound calls. In this example, create a speech-to-speech agent using OpenAI's realtime API. The same example can also be used with VoicePipelineAgent.

Create a template outbound caller agent:

lk app create --template=outbound-caller-python

Follow the instructions in the command output. Enter the outbound trunk ID from Step 2.

Understanding the code

There are a few key points of note in this example detailed in the following sections.

Set `agent_name` for explicit dispatch

if __name__ == "__main__":
    cli.run_app(
        WorkerOptions(
            entrypoint_fnc=entrypoint,
            # giving this agent a name will allow us to dispatch it via API
            # automatic dispatch is disabled when `agent_name` is set
            agent_name="outbound-caller",
        )
    )

By setting the agent_name field, you can use AgentDispatchService to create a dispatch for this agent. To learn more, see agent dispatch.

Dialing the user

async def entrypoint(ctx: JobContext):
    global _default_instructions
    logger.info(f"connecting to room {ctx.room.name}")
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)

    user_identity = "phone_user"
    # the phone number to dial is provided in the job metadata
    phone_number = ctx.job.metadata
    logger.info(f"dialing {phone_number} to room {ctx.room.name}")

    # look up the user's phone number and appointment details
    instructions = _default_instructions + "The customer's name is Jayden. His appointment is next Tuesday at 3pm."

    # `create_sip_participant` starts dialing the user
    await ctx.api.sip.create_sip_participant(api.CreateSIPParticipantRequest(
        room_name=ctx.room.name,
        sip_trunk_id=outbound_trunk_id,
        sip_call_to=phone_number,
        participant_identity=user_identity,
    ))

    # a participant is created as soon as we start dialing
    participant = await ctx.wait_for_participant(identity=user_identity)

    # start either VoicePipelineAgent or MultimodalAgent
    #run_voice_pipeline_agent(ctx, participant, instructions)
    run_multimodal_agent(ctx, participant, instructions)

Once dispatched, the agent first connects to the room and then dials the user. The agent receives instructions on which user to call from the job metadata set during dispatch. In this example, the user’s phone number is included in the job metadata (details below).

After determining which user to call, you can load additional information about the user from your own database.

Finally, calling create_sip_participant initiates dialing the user.

Monitoring dialing status

start_time = perf_counter()
while perf_counter() - start_time < 30:
    call_status = participant.attributes.get("sip.callStatus")
    if call_status == "active":
        logger.info("user has picked up")
        return
    elif call_status == "automation":
        # if DTMF is used in the `sip_call_to` number, typically used to dial
        # an extension or enter a PIN.
        # during DTMF dialing, the participant will be in the "automation" state
        pass
    elif call_status == "hangup":
        # user hung up, we'll exit the job
        logger.info("user hung up, exiting job")
        break
    await asyncio.sleep(0.1)

logger.info("session timed out, exiting job")
ctx.shutdown()

Once create_sip_participant is called, LiveKit begins dialing the user. The method returns immediately after dialing is initiated and does not wait for the user to answer.

Optionally, you can check the sip.callStatus attribute to monitor dialing status. When the user answers, this attribute updates to active.

Handle actions with function calling

Use @llm.ai_callable() to prepare functions for the LLM to use as tools.

In this example, the following actions are handled:

Detecting voicemail
Looking up availability
Confirming the appointment
Detecting intent to end the call

class CallActions(llm.FunctionContext):
    def __init__(self, *, api: api.LiveKitAPI, participant: rtc.RemoteParticipant, room: rtc.Room):
        super().__init__()

        self.api = api
        self.participant = participant
        self.room = room

    async def hangup(self):
        try:
            await self.api.room.remove_participant(api.RoomParticipantIdentity(
                room=self.room.name,
                identity=self.participant.identity,
            ))
        except Exception as e:
            # it's possible that the user has already hung up, this error can be ignored
            logger.info(f"received error while ending call: {e}")

    @llm.ai_callable()
    async def end_call(self):
        """Called when the user wants to end the call"""
        logger.info(f"ending the call for {self.participant.identity}")
        await self.hangup()

    @llm.ai_callable()
    async def look_up_availability(self,
                                   date: Annotated[str, "The date of the appointment to check availability for"],
                                   ):
        """Called when the user asks about alternative appointment availability"""
        logger.info(f"looking up availability for {self.participant.identity} on {date}")
        asyncio.sleep(3)
        return json.dumps({
            "available_times": ["1pm", "2pm", "3pm"],
        })

    @llm.ai_callable()
    async def confirm_appointment(self,
                                  date: Annotated[str, "date of the appointment"],
                                  time: Annotated[str, "time of the appointment"],
                                  ):
        """Called when the user confirms their appointment on a specific date. Use this tool only when they are certain about the date and time."""
        logger.info(f"confirming appointment for {self.participant.identity} on {date} at {time}")
        return "reservation confirmed"

    @llm.ai_callable()
    async def detected_answering_machine(self):
        """Called when the call reaches voicemail. Use this tool AFTER you hear the voicemail greeting"""
        logger.info(f"detected answering machine for {self.participant.identity}")
        await self.hangup()

...
agent = MultimodalAgent(
    model=model,
    fnc_ctx=CallActions(api=ctx.api, participant=participant, room=ctx.room),
)

Step 4: Creating a dispatch

Once your agent is up and running, you can test it by having it make a phone call.

Use LiveKit CLI to create a dispatch. Replace +15105550100 with the phone number you want to call.