python-sdks
python-sdks copied to clipboard
livekit
Updating dispatch rules and trunks fails with "no handler for path"
Important Info: I am using a self hosted livekit server 1.8.3 (which works with playground and registering the worker etc.) and livekit SIP server v0.10 (which points to the latest digest here). The script works against LiveKit Cloud.
Not sure if this is an expected error.
I am using livekit-api==1.0.2 via this dependency:
"livekit-agents[deepgram,openai,cartesia,silero,turn-detector]~=1.0"
I am trying to write a script which updates the sip rules if they exist already, but this fails with these errors
Connecting to the livekit server via CLI, and creating all the entries initially with this script works as expected.
Traceback (most recent call last):
File "/Users/ademfrenk/github.com/enam-co/mono/services/agent_worker/scripts/sip_inbound_init.py", line 107, in <module>
asyncio.run(main())
File "/Users/ademfrenk/.pyenv/versions/3.12.9/lib/python3.12/asyncio/runners.py", line 195, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/Users/ademfrenk/.pyenv/versions/3.12.9/lib/python3.12/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/ademfrenk/.pyenv/versions/3.12.9/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^
File "/Users/ademfrenk/github.com/enam-co/mono/services/agent_worker/scripts/sip_inbound_init.py", line 101, in main
await update_or_create_dispatch_rule(lk_api, trunk)
File "/Users/ademfrenk/github.com/enam-co/mono/services/agent_worker/scripts/sip_inbound_init.py", line 84, in update_or_create_dispatch_rule
dispatch = await lk_api.sip.update_sip_dispatch_rule(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/ademfrenk/github.com/enam-co/mono/services/agent_worker/.venv/lib/python3.12/site-packages/livekit/api/sip_service.py", line 310, in update_sip_dispatch_rule
return await self._client.request(
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/ademfrenk/github.com/enam-co/mono/services/agent_worker/.venv/lib/python3.12/site-packages/livekit/api/twirp_client.py", line 129, in request
raise TwirpError(
livekit.api.twirp_client.TwirpError: TwirpError(code=bad_route, message=no handler for path "/twirp/livekit.SIP/UpdateSIPDispatchRule", status=404, metadata={'twirp_invalid_route': 'POST /twirp/livekit.SIP/UpdateSIPDispatchRule'})
Same for Trunk with
Traceback (most recent call last):
File "/Users/ademfrenk/github.com/enam-co/mono/services/agent_worker/scripts/sip_inbound_init.py", line 109, in <module>
asyncio.run(main())
File "/Users/ademfrenk/.pyenv/versions/3.12.9/lib/python3.12/asyncio/runners.py", line 195, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/Users/ademfrenk/.pyenv/versions/3.12.9/lib/python3.12/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/ademfrenk/.pyenv/versions/3.12.9/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^
File "/Users/ademfrenk/github.com/enam-co/mono/services/agent_worker/scripts/sip_inbound_init.py", line 102, in main
trunk = await update_or_create_trunk(lk_api)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/ademfrenk/github.com/enam-co/mono/services/agent_worker/scripts/sip_inbound_init.py", line 42, in update_or_create_trunk
trunk_result = await lk_api.sip.update_sip_inbound_trunk(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/ademfrenk/github.com/enam-co/mono/services/agent_worker/.venv/lib/python3.12/site-packages/livekit/api/sip_service.py", line 90, in update_sip_inbound_trunk
return await self._client.request(
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/ademfrenk/github.com/enam-co/mono/services/agent_worker/.venv/lib/python3.12/site-packages/livekit/api/twirp_client.py", line 129, in request
raise TwirpError(
livekit.api.twirp_client.TwirpError: TwirpError(code=bad_route, message=no handler for path "/twirp/livekit.SIP/UpdateSIPInboundTrunk", status=404, metadata={'twirp_invalid_route': 'POST /twirp/livekit.SIP/UpdateSIPInboundTrunk'})
This is the script:
import asyncio
import os
from dotenv import load_dotenv
from livekit import api
from livekit.protocol.sip import ListSIPDispatchRuleRequest, ListSIPInboundTrunkRequest
load_dotenv(dotenv_path=".agent_worker.env")
LIVEKIT_INBOUND_TENANT_ID = os.environ.get("LIVEKIT_INBOUND_TENANT_ID")
if not LIVEKIT_INBOUND_TENANT_ID:
raise ValueError(
"LIVEKIT_INBOUND_TENANT_ID must be set to identify which tenant this phone number belongs to"
)
NUMBER = os.environ.get("LIVEKIT_TWILIO_INBOUND_NUMBER")
if not NUMBER:
raise ValueError("LIVEKIT_TWILIO_INBOUND_NUMBER is not set")
AGENT_NAME = os.environ.get("LIVEKIT_INBOUND_AGENT_NAME")
if not AGENT_NAME:
raise ValueError("LIVEKIT_INBOUND_AGENT_NAME is not set")
async def update_or_create_trunk(lk_api: api.LiveKitAPI) -> api.SIPInboundTrunkInfo:
trunk = api.SIPInboundTrunkInfo(
name="Inbound dev trunk",
numbers=[NUMBER],
krisp_enabled=False, # Noise cancellation
# allowed_numbers = ["+13105550100"] # Optional limitation to specific numbers
)
# Update trunk for this number, if it already exists
resp = await lk_api.sip.list_sip_inbound_trunk(ListSIPInboundTrunkRequest())
for _trunk in resp.items:
if NUMBER in _trunk.numbers and _trunk.name == trunk.name:
trunk_result = await lk_api.sip.update_sip_inbound_trunk(
trunk_id=_trunk.sip_trunk_id,
trunk=trunk,
)
print(f"Updated trunk for number {NUMBER}:\n{trunk_result}")
return trunk_result
# Create trunk if no trunk for number exists yet
request = api.CreateSIPInboundTrunkRequest(trunk=trunk)
trunk_result = await lk_api.sip.create_sip_inbound_trunk(request)
print(f"Created trunk for number {NUMBER}:\n{trunk_result}")
return trunk_result
async def update_or_create_dispatch_rule(
lk_api: api.LiveKitAPI, trunk: api.SIPInboundTrunkInfo
):
rule = api.SIPDispatchRule(
dispatch_rule_individual=api.SIPDispatchRuleIndividual(
room_prefix="call-",
)
)
room_config = api.RoomConfiguration(
agents=[
api.RoomAgentDispatch(
agent_name=AGENT_NAME, # Naming an agent will disable automatic dispatch, so this is explicitly needed
)
]
)
info_args = dict(
name="Inbound dev dispatch rule",
trunk_ids=[trunk.sip_trunk_id],
# inbound_numbers=[NUMBER], # NOTE: This is broken for some reason. If set, inbound calls won't work.
rule=rule,
room_config=room_config,
attributes={
"tenant_id": LIVEKIT_INBOUND_TENANT_ID
}, # Set attributes on dispatch rule
)
# Update existing dispatch rule if found
resp = await lk_api.sip.list_sip_dispatch_rule(ListSIPDispatchRuleRequest())
for _rule in resp.items:
if _rule.name == info_args["name"]:
dispatch = await lk_api.sip.update_sip_dispatch_rule(
rule_id=_rule.sip_dispatch_rule_id,
rule=api.SIPDispatchRuleInfo(**info_args),
)
print(f"Updated dispatch rule:\n{dispatch}")
return
# Create new dispatch rule if none exists
request = api.CreateSIPDispatchRuleRequest(**info_args)
dispatch = await lk_api.sip.create_sip_dispatch_rule(request)
print("Created dispatch rule:\n", dispatch)
async def main():
lk_api = api.LiveKitAPI()
try:
trunk = await update_or_create_trunk(lk_api)
await update_or_create_dispatch_rule(lk_api, trunk)
finally:
await lk_api.aclose()
if __name__ == "__main__":
asyncio.run(main())
