https://livekit.io logo
Join Slack
Powered by
Can you give the json format for creating inbound ...
# ask-ai
m
Can you give the json format for creating inbound trunk to serve locally?
t
Thanks for reaching out, Rahib. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Hi there! To help you create an inbound trunk that serves locally, here’s a simple JSON format you can use:
Copy code
{
  "trunk": {
    "name": "My inbound trunk",
    "numbers": [
      "+15105550100"
    ]
  }
}
There are also some additional options you might find useful: • `allowed_addresses`: This lets you restrict inbound calls to specific IPs or CIDR blocks for extra security. • `allowed_numbers`: Use this to limit inbound calls to particular caller numbers. • 
auth_username
and `auth_password`: These are your credentials if your SIP provider requires authentication. • `krisp_enabled`: Enable this to reduce background noise during calls. • `metadata`: Add custom metadata for participants joining through this trunk. Here’s an example including some of those restrictions:
Copy code
{
  "trunk": {
    "name": "My trunk",
    "numbers": [
      "+15105550100"
    ],
    "allowedNumbers": [
      "+13105550100",
      "+17145550100"
    ]
  }
}
If your SIP provider supports username/password authentication, you can include it like this:
Copy code
{
  "trunk": {
    "name": "My inbound trunk",
    "auth_username": "your_username",
    "auth_password": "your_password"
  }
}
Just a quick reminder: make sure all phone numbers are in E.164 format (for example, 
+15105550100
). Once you’ve prepared your JSON, you can easily create the trunk using the LiveKit CLI with this command:
Copy code
lk sip inbound create inbound-trunk.json
Alternatively, you can use the HTTP API endpoint 
/twirp/livekit.SIP/CreateSIPInboundTrunk
by sending the JSON body along with your authorization token. If you have any questions or need further help, feel free to ask—I’m here to support you! Sources: Accepting incoming calls | LiveKit Docs | Accepting incoming calls | LiveKit Docs | SIP inbound trunk | LiveKit Docs | SIP APIs | LiveKit Docs
m
Does inbound trunk need to match anything from dispatch rules?
t
I'm glad you asked! To successfully accept and process inbound SIP calls in LiveKit, an inbound trunk needs to match a dispatch rule. When an inbound call comes in, LiveKit first authenticates the trunk and then looks for a dispatch rule that matches to decide how to handle the call. If no matching dispatch rule is found, the call will unfortunately be rejected with a "Does not match Trunks or Dispatch Rules" response. Here are some important things to keep in mind: • The inbound trunk specifies which phone numbers and IP addresses are allowed to send calls to LiveKit. • The dispatch rule determines how those calls are routed, like placing callers into specific rooms or creating new rooms. • Dispatch rules can be linked to specific trunks using 
trunkIds
, or they can serve as wildcard rules that apply more broadly. So, both the inbound trunk and a matching dispatch rule are essential to have your inbound call setup working smoothly. If you need any help setting this up, feel free to ask—I’m here to help! Sources: SIP dispatch rule | LiveKit Docs | pkg/service/ioservice_sip.go | Accepting inbound calls | LiveKit Docs | pkg/sip/inbound.go | SIP inbound trunk | LiveKit Docs
m
2025-09-17 171130,047 - ERROR livekit - livekit_ffi:serverroom149livekit ffiserver:room - audio filter cannot be enabled: LiveKit Cloud is required {"room": "telephony-voice-assistant__1200_CS4YALh9fu3m", "pid": 65556, "job_id": "AJ_LSsKSGPGumHZ"}
t
Hi there! I wanted to explain why you might be seeing the error "audio filter cannot be enabled: LiveKit Cloud is required." This happens because some of the enhanced audio filtering features, like Krisp noise cancellation, are only available when using LiveKit Cloud. Unfortunately, these features aren’t supported on self-hosted LiveKit server instances. This is because the licensed models from Krisp are exclusively accessible through LiveKit Cloud. So, when you’re running a self-hosted setup, the system automatically disables the audio filter and logs this message, allowing your RTC session to keep running smoothly without it. If enhanced audio filtering is something you’d like to use, migrating to LiveKit Cloud would be the way to go. Please feel free to reach out if you’d like help with the migration or if you have any other questions—I’m here to support you! Sources: Enhanced noise cancellation | LiveKit Docs | Operational Guidance | livekit/livekit_composite | DeepWiki | Development Tools | livekit/livekit_composite | DeepWiki | webrtc-sys/include/livekit/apm.h | README.md
m
how do i log the participants inside the room, once the call is connected with the agent. Or basically see how many participatns are there
Copy code
import asyncio
import logging
import os

from dotenv import load_dotenv
from livekit.agents import (
    NOT_GIVEN,
    Agent,
    AgentFalseInterruptionEvent,
    AgentSession,
    JobContext,
    JobProcess,
    MetricsCollectedEvent,
    RoomInputOptions,
    RunContext,
    WorkerOptions,
    cli,
    metrics,
)
from livekit import rtc
from livekit.agents.llm import function_tool
from livekit.agents.metrics import EOUMetrics, LLMMetrics, STTMetrics, TTSMetrics
from livekit.plugins import cartesia, deepgram, noise_cancellation, openai, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel

logger = logging.getLogger("agent")

load_dotenv(".env.local")


class MetricsAgent(Agent):
    def __init__(self) -> None:
        llm = openai.LLM.with_ollama(
            model="qwen2.5:3b", base_url="<http://localhost:11434/v1>", temperature=0.3
        )
        stt = deepgram.STT(
            model="nova-3",
            language="multi",
            api_key=os.getenv("DEEPGRAM_API_KEY")
        )
        tts = cartesia.TTS(
            voice="6f84f4b8-58a2-430c-8c79-688dad597532",
            api_key=os.getenv("CARTESIA_API_KEY")
        )

        silero_vad = silero.VAD.load()

        super().__init__(
            instructions="You are a helpful assistant communicating via voice",
            stt=stt,
            llm=llm,
            tts=tts,
            vad=silero_vad,
        )

        def llm_metrics_wrapper(metrics: LLMMetrics):
            asyncio.create_task(self.on_llm_metrics_collected(metrics))

        llm.on("metrics_collected", llm_metrics_wrapper)

        def stt_metrics_wrapper(metrics: STTMetrics):
            asyncio.create_task(self.on_stt_metrics_collected(metrics))

        stt.on("metrics_collected", stt_metrics_wrapper)

        def eou_metrics_wrapper(metrics: EOUMetrics):
            asyncio.create_task(self.on_eou_metrics_collected(metrics))

        stt.on("eou_metrics_collected", eou_metrics_wrapper)

        def tts_metrics_wrapper(metrics: TTSMetrics):
            asyncio.create_task(self.on_tts_metrics_collected(metrics))

        tts.on("metrics_collected", tts_metrics_wrapper)

    async def on_llm_metrics_collected(self, metrics: LLMMetrics) -> None:
        print("\n--- LLM Metrics ---")
        print(f"Prompt Tokens: {metrics.prompt_tokens}")
        print(f"Completion Tokens: {metrics.completion_tokens}")
        print(f"Tokens per second: {metrics.tokens_per_second:.4f}")
        print(f"TTFT: {metrics.ttft:.4f}s")
        print("------------------\n")

    async def on_stt_metrics_collected(self, metrics: STTMetrics) -> None:
        print("\n--- STT Metrics ---")
        print(f"Duration: {metrics.duration:.4f}s")
        print(f"Audio Duration: {metrics.audio_duration:.4f}s")
        print(f"Streamed: {'Yes' if metrics.streamed else 'No'}")
        print("------------------\n")

    async def on_eou_metrics_collected(self, metrics: EOUMetrics) -> None:
        print("\n--- End of Utterance Metrics ---")
        print(f"End of Utterance Delay: {metrics.end_of_utterance_delay:.4f}s")
        print(f"Transcription Delay: {metrics.transcription_delay:.4f}s")
        print("--------------------------------\n")

    async def on_tts_metrics_collected(self, metrics: TTSMetrics) -> None:
        print("\n--- TTS Metrics ---")
        print(f"TTFB: {metrics.ttfb:.4f}s")
        print(f"Duration: {metrics.duration:.4f}s")
        print(f"Audio Duration: {metrics.audio_duration:.4f}s")
        print(f"Streamed: {'Yes' if metrics.streamed else 'No'}")
        print("------------------\n")


class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""You are a helpful voice AI assistant.
            You eagerly assist users with their questions by providing information from your extensive knowledge.
            Your responses are concise, to the point, and without any complex formatting or punctuation including emojis, asterisks, or other symbols.
            You are curious, friendly, and have a sense of humor.""",
        )

    # all functions annotated with @function_tool will be passed to the LLM when this
    # agent is active
    @function_tool
    async def lookup_weather(self, context: RunContext, location: str):
        """Use this tool to look up current weather information in the given location.

        If the location is not supported by the weather service, the tool will indicate this. You must tell the user the location's weather is unavailable.

        Args:
            location: The location to look up weather information for (e.g. city name)
        """

        <http://logger.info|logger.info>(f"Looking up weather for {location}")

        return "sunny with a temperature of 70 degrees."


def prewarm(proc: JobProcess):
    proc.userdata["vad"] = silero.VAD.load()


async def entrypoint(ctx: JobContext):
    # Logging setup
    # Add any other context you want in all log entries here
    ctx.log_context_fields = {
        "room": ctx.room.name,
    }


    # # Add a virtual avatar to the session, if desired
    # # For other providers, see <https://docs.livekit.io/agents/integrations/avatar/>
    # avatar = hedra.AvatarSession(
    #   avatar_id="...",  # See <https://docs.livekit.io/agents/integrations/avatar/hedra>
    # )
    # # Start the avatar and wait for it to join
    # await avatar.start(session, room=ctx.room)
    ctx.log_context_fields = {
        "room_name": ctx.room.name,
    }
    session - AgentSession()
    # Start the session, which initializes the voice pipeline and warms up the models
    await session.start(
        agent=MetricsAgent(),
        room=ctx.room,
        room_input_options=RoomInputOptions(
            # LiveKit Cloud enhanced noise cancellation
            # - If self-hosting, omit this parameter
            # - For telephony applications, use `BVCTelephony` for best results
            # noise_cancellation=noise_cancellation.BVC(),
        ),
    )

    # Join the room and connect to the user
    await ctx.connect()


if __name__ == "__main__":
    cli.run_app(
        WorkerOptions(
            entrypoint_fnc=entrypoint,
            prewarm_fnc=prewarm,
            agent_name="telephony-voice-assistant",
        )
    )
t
Thanks for reaching out! Just to make sure I understand correctly, are you hoping to log participant details like names and IDs, or are you looking to simply count how many participants are in the room? Also, could you let me know if you're using LiveKit's Python SDK or another way to interact with the room? I’m here to help you through this!
m
I run my file using uv run
Copy code
if __name__ == "__main__":
    cli.run_app(
        WorkerOptions(
            entrypoint_fnc=entrypoint,
            prewarm_fnc=prewarm,
            agent_name="telephony-voice-assistant",
        )
    )
I want to see if the participant is actually connected or not. The SIP server and livekit server are running locally in my docker container. Basically kind of log if partiicpatn and the agent are in the same room when the call is made.
t
I'm glad you're looking to check if participants (like the agent and caller) are in the same room and want to log their presence. Here's a friendly guide using the LiveKit Python SDK that should help you out: 1. Access Current Participants Once you’re connected to the room, you can easily list all connected participants using the 
remote_participants
property:
Copy code
for identity, participant in room.remote_participants.items():
    print(f"Participant connected: {participant.identity} (SID: {participant.sid})")
2. Listen for Participant Events It’s really helpful to register event listeners so you can keep track of when participants join or leave the room:
Copy code
@room.on("participant_connected")
def on_participant_connected(participant):
    print(f"Participant joined: {participant.identity}")

@room.on("participant_disconnected")
def on_participant_disconnected(participant):
    print(f"Participant left: {participant.identity}")
3. Verify Room Connection Before you try to access participants, just make sure the room is connected. This way, you avoid any surprises:
Copy code
if room.isconnected():
    print("Room is connected, checking participants...")
    for p in room.remote_participants.values():
        print(f"- {p.identity}")
else:
    print("Not connected to room")
4. Full Example
Copy code
import asyncio
from livekit import rtc

async def main():
    room = rtc.Room()

    @room.on("participant_connected")
    def on_participant_connected(participant):
        print(f"Participant connected: {participant.identity}")

    @room.on("participant_disconnected")
    def on_participant_disconnected(participant):
        print(f"Participant disconnected: {participant.identity}")

    await room.connect("your-room-url", "your-access-token")

    # List existing participants after connecting
    print("Current participants:")
    for p in room.remote_participants.values():
        print(f"- {p.identity}")

asyncio.run(main())
If you’d like, I’m here to help you integrate this into your existing code or answer any other questions you might have. Just let me know! Sources: livekit-rtc/livekit/rtc/_proto/room_pb2.pyi | Rooms, participants, and tracks | LiveKit Docs
m
Yeah what would be the best way to integrate in existing example
t
To help you integrate participant logging into your existing example, you can update the 
entrypoint
function by adding event listeners for participant connections and disconnections. Here’s a friendly guide on how to do it:
Copy code
async def entrypoint(ctx: JobContext):
    # Logging setup
    ctx.log_context_fields = {
        "room_name": ctx.room.name,
    }

    session = AgentSession()

    # Add event listeners for participant connections and disconnections
    @ctx.room.on("participant_connected")
    def on_participant_connected(participant):
        print(f"Participant connected: {participant.identity}")

    @ctx.room.on("participant_disconnected")
    def on_participant_disconnected(participant):
        print(f"Participant disconnected: {participant.identity}")

    # Start the session and connect to the room
    await session.start(
        agent=MetricsAgent(),
        room=ctx.room,
        room_input_options=RoomInputOptions(
            # Add any additional options here
        ),
    )

    # Log existing participants after connecting
    print("Current participants:")
    for identity, participant in ctx.room.remote_participants.items():
        print(f"- {participant.identity}")

    # Join the room and connect to the user
    await ctx.connect()
This approach will help you keep track of participants already in the room when the session starts, and it will also log dynamically whenever participants join or leave. If you have any questions or need a hand with anything else, please don’t hesitate to reach out—I’m here to help!
m
I never saw the log for partiicpant connected. But i can see it for participant disconnected. Participant disconnected: sip_testing-user-cmetric and Current participants: - sip_testing-user-cmetric only shows the user that called and not the agent? does this also supposed to log agent as well? Am i doing something wrong?
t
It sounds like the issue you're facing might be related to how LiveKit manages participant kinds and filters out agent participants by default. Let me walk you through the important details: • In LiveKit, participants are grouped by kinds, such as 
PARTICIPANT_KIND_SIP
, 
PARTICIPANT_KIND_STANDARD
, and 
PARTICIPANT_KIND_AGENT
. • By default, the 
RoomInputOptions
configuration only includes 
PARTICIPANT_KIND_SIP
and 
PARTICIPANT_KIND_STANDARD
. So, agents (classified as 
PARTICIPANT_KIND_AGENT
) are left out unless you explicitly add them to the 
participant_kinds
parameter. • Also, LiveKit may filter out participants with the 
ATTRIBUTE_PUBLISH_ON_BEHALF
attribute to avoid agents processing their own streams. To make sure your agent is properly logged as a participant, you can: 1. Update your 
RoomInputOptions
to include 
PARTICIPANT_KIND_AGENT
in the 
participant_kinds
list. 2. Review your code or middleware to see if there's any filtering that excludes participants where 
kind == PARTICIPANT_KIND_AGENT
. 3. Double-check that your event handlers aren’t set to ignore agent participants. If you'd like, I’m here to help you adjust your configuration or troubleshoot further—just let me know! Sources: livekit-agents/livekit/agents/job.py | agent/game_host.py
m
Since i am running livekit server and sip server locally could that be an issue? Below is how i start my session
Copy code
await session.start(
        agent=MetricsAgent(),
        room=ctx.room,
        room_input_options=RoomInputOptions(
            # LiveKit Cloud enhanced noise cancellation
            # - If self-hosting, omit this parameter
            # - For telephony applications, use `BVCTelephony` for best results
            # noise_cancellation=noise_cancellation.BVC(),
        ),
    )
t
I completely understand how important it is to get things working smoothly! Running the LiveKit server and SIP server locally doesn’t actually change how participant connection events or 
RoomInputOptions
behave. These features are built right into the LiveKit protocol and SDKs, so they work the same whether your servers are running locally or remotely. Events like 
participant_connected
and 
participant_disconnected
happen based on the participants’ actual status in the room. Likewise, the settings you choose in 
RoomInputOptions
, such as which participant kinds are accepted, are handled at the SDK level and aren’t affected by where your server is hosted. If things aren’t behaving as expected, it might be worth double-checking your 
RoomInputOptions
configuration or any filtering logic in your code. I’m here to help if you want to go over your setup together! Sources: pkg/sip/room.go | livekit-agents/livekit/agents/voice/room_io/room_io.py | agents/src/voice/room_io/room_io.ts | Running LiveKit locally | LiveKit Docs | Self-hosted SIP server | LiveKit Docs
m
so give me config on how to set them ? and what is the default in that case.
t
Thanks for reaching out! By default, the 
RoomInputOptions
in LiveKit accepts participants of kinds 
PARTICIPANT_KIND_SIP
and 
PARTICIPANT_KIND_STANDARD
, but it doesn’t include 
PARTICIPANT_KIND_AGENT
unless you specifically add it. To make sure agents are part of your setup, you’ll want to update the 
participant_kinds
parameter. Here’s a friendly example of how you can configure 
RoomInputOptions
to include all participant kinds:
Copy code
room_input_options = RoomInputOptions(
    participant_kinds=[
        rtc.ParticipantKind.PARTICIPANT_KIND_SIP,
        rtc.ParticipantKind.PARTICIPANT_KIND_STANDARD,
        rtc.ParticipantKind.PARTICIPANT_KIND_AGENT
    ]
)
Then, just pass this configuration when you start your session:
Copy code
await session.start(
    agent=MetricsAgent(),
    room=ctx.room,
    room_input_options=room_input_options
)
This way, all participant kinds, including agents, will be properly processed and logged in the room. If you have any questions or need a hand with this, feel free to ask—I’m here to help!
Open in Slack
PreviousNext
Powered by