dry-area-57176
07/24/2025, 8:56 AMtall-airline-16756
07/24/2025, 8:56 AMtall-airline-16756
07/24/2025, 9:00 AMexport 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:
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:
{
"rule": {
"dispatchRuleIndividual": {
"roomPrefix": "call-"
}
},
"roomConfig": {
"agents": [
{
"agentName": "sip-agent"
}
]
}
}
Apply the dispatch rule:
lk sip dispatch create dispatch-rule.json
4. Outbound Call Example:
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