i have a self hosted livekit sip and livekit serve...
# ask-ai
d
i have a self hosted livekit sip and livekit server instance. How can i connect a custom sip account to livekit so when i call the number i can talk directly to ai agent. Give me a python example
t
Thanks for reaching out, Niklas. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
To connect a custom SIP account to your self-hosted LiveKit server and enable calls to an AI agent, you can follow these steps: 1. Set Up Your Environment: • Ensure your LiveKit server is properly configured and running. • Set the required environment variables for your LiveKit instance:
Copy code
export LIVEKIT_URL=<your_livekit_server_url>
export LIVEKIT_API_KEY=<your_api_key>
export LIVEKIT_API_SECRET=<your_api_secret>
2. Python Example for SIP Integration: Below is a Python example that demonstrates how to integrate SIP with an AI agent:
Copy code
import asyncio
import json
from livekit import api
from livekit.agents import (
    AgentSession,
    Agent,
    JobContext,
    cli,
    WorkerOptions,
    function_tool,
    RunContext,
)
from livekit.plugins import deepgram, openai, cartesia, silero

class SIPAgent(Agent):
    def __init__(self, **kwargs):
        super().__init__(
            instructions="""
            You are a helpful assistant communicating through voice.
            You handle phone calls and assist with various inquiries.
            Be polite and professional at all times.
            """,
        )
        self.participant = None

    def set_participant(self, participant):
        self.participant = participant

    @function_tool()
    async def transfer_call(self, ctx: RunContext, phone_number: str):
        """Transfer the call to another number"""
        if not self.participant:
            return "Cannot transfer call - no participant found"
        
        job_ctx = ctx.job_context
        try:
            await job_ctx.api.sip.transfer_sip_participant(
                api.TransferSIPParticipantRequest(
                    room_name=job_ctx.room.name,
                    participant_identity=self.participant.identity,
                    transfer_to=f"tel:{phone_number}",
                )
            )
            return f"Call transferred to {phone_number}"
        except Exception as e:
            return f"Transfer failed: {e}"

    @function_tool()
    async def end_call(self, ctx: RunContext):
        """End the current call"""
        job_ctx = ctx.job_context
        await job_ctx.api.room.delete_room(
            api.DeleteRoomRequest(room=job_ctx.room.name)
        )

async def entrypoint(ctx: JobContext):
    await ctx.connect()
    
    session = AgentSession(
        stt=deepgram.STT(),
        tts=cartesia.TTS(),
        llm=openai.LLM(model="gpt-4o"),
        vad=silero.VAD.load()
    )
    
    agent = SIPAgent()
    
    if hasattr(ctx.job, 'metadata') and ctx.job.metadata:
        dial_info = json.loads(ctx.job.metadata)
        phone_number = dial_info.get("phone_number")
        
        if phone_number:
            try:
                await ctx.api.sip.create_sip_participant(
                    api.CreateSIPParticipantRequest(
                        room_name=ctx.room.name,
                        sip_trunk_id="your_trunk_id",
                        sip_call_to=phone_number,
                        participant_identity=phone_number,
                        wait_until_answered=True,
                    )
                )
                print("Call connected successfully")
            except api.TwirpError as e:
                print(f"Error creating SIP participant: {e.message}")
                ctx.shutdown()
                return
    
    await session.start(agent=agent, room=ctx.room)
    
    if ctx.room.remote_participants:
        participant = list(ctx.room.remote_participants.values())[0]
        agent.set_participant(participant)
    
    if not (hasattr(ctx.job, 'metadata') and ctx.job.metadata):
        await session.generate_reply(
            instructions="Greet the user and offer your assistance."
        )

if __name__ == "__main__":
    cli.run_app(
        WorkerOptions(
            entrypoint_fnc=entrypoint,
            agent_name="sip-agent",
        ),
    )
3. Configure Dispatch Rules: • Create dispatch rules to route inbound SIP calls to LiveKit rooms:
Copy code
{
  "rule": {
    "dispatchRuleIndividual": {
      "roomPrefix": "call-"
    }
  },
  "roomConfig": {
    "agents": [
      {
        "agentName": "sip-agent"
      }
    ]
  }
}
Apply the dispatch rule:
Copy code
lk sip dispatch create dispatch-rule.json
4. Outbound Call Example:
Copy code
lk dispatch create \
  --new-room \
  --agent-name sip-agent \
  --metadata '{"phone_number": "+1234567890"}'
Note: For custom SIP account integration, you may need to configure your SIP server to handle custom account registrations and route calls to LiveKit. This setup may require additional customization beyond the standard trunk provider approach. Sources: Integrating with Telephony | LiveKit Docs | Making calls using SIP | LiveKit Docs | Accepting incoming calls | LiveKit Docs | Agents telephony integration | LiveKit Docs | README.md