Hi, how can i check what is happening/happend with call coming from SIP provider, for which room is got created but no agent got assigned?

Thanks for reaching out, Harveer. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

I am not passing agent_name WorkerOptions so it make automatic dispatch but we have agents param set to one agent in liast of agent with metadata

Hey there! When you specify agents in metadata, LiveKit switches to explicit dispatch instead of automatic dispatch. Here’s a quick rundown of how that works: 1. Explicit Dispatch: Once agents are specified, automatic dispatch takes a backseat. You’ll need to assign the agent to a room yourself, either through the API or when generating the token. 2. Metadata Usage: You can pass metadata during dispatch—think user IDs or job-specific info. This metadata is then available in the agent's

JobContext

. 3. Room Configuration: You can set up a participant’s access token to dispatch agents as soon as they connect, including any metadata for the agent. 4. Room Creation: Don’t worry if the room doesn’t exist yet—LiveKit will create it automatically during dispatch. Since you’re working with metadata, just make sure the agent is explicitly dispatched to the room. If you want, I’m here to help with more details or examples whenever you need them! Sources: Dispatching agents | LiveKit Docs | packages/livekit-server-sdk/src/AgentDispatchClient.ts | Agent dispatch | LiveKit Docs

can give me example how it will looks like from entrypoint prespective?

Here’s a friendly example to show you how to handle metadata in a LiveKit agent’s entrypoint:

import json

async def entrypoint(ctx: JobContext):
    # First, let's parse the job metadata
    metadata = json.loads(ctx.job.metadata)

    # Then, we grab some key details
    user_id = metadata.get("user_id")
    user_name = metadata.get("user_name")
    phone_number = metadata.get("phone_number")

    # Now, we use that info to set up the agent
    session = AgentSession()
    agent = YourAgent(
        instructions=f"Hello {user_name}!",
        context={"user_id": user_id}
    )

    # Finally, connect the agent to the room
    await ctx.connect()

Here’s a quick rundown: • We start by parsing the metadata from

ctx.job.metadata

. • Then, we pull out important bits like

user_id

and

user_name

to help guide the agent’s behavior. • And lastly, the agent connects to the room with

ctx.connect()

. If you want me to walk you through any part of this or need some more examples, just let me know—I’m happy to help!

okay so I am getting SIP call to SIP server, then server is creating room now i want to agent that room with dispatch rule metadata, so i can identify user and tenent details to process it

Hey there! If you want to assign agents to SIP-created rooms using dispatch rules with metadata, here’s a simple way to do it: 1. Create a SIP Dispatch Rule: This rule will manage incoming SIP calls. For example:

{
  "name": "My SIP Dispatch Rule",
  "rule": {
    "dispatchRuleIndividual": {
      "roomPrefix": "sip-call-"
    }
  },
  "room_config": {
    "agents": [
      {
        "agentName": "sip-agent",
        "metadata": "{\"source\": \"sip\", \"call_id\": \"12345\"}"
      }
    ]
  }
}

2. Set Custom Metadata: Use the

metadata

field to add user and tenant details. This helps with processing the calls smoothly. 3. Apply the Dispatch Rule: Just run this command with the LiveKit CLI to apply your rule to a SIP trunk:

lk sip dispatch create dispatch-rule.json

4. Handle Metadata in Agents: In your agent’s entrypoint, you can grab the metadata like this:

async def entrypoint(ctx: JobContext):
    <http://logger.info|logger.info>(f"Job metadata: {ctx.job.metadata}")
    <http://logger.info|logger.info>(f"Source: {ctx.job.metadata.get('source')}")
    # Process call based on metadata

And that’s it! This setup makes sure your agents get assigned to SIP-created rooms with all the metadata they need. If you have any questions or want me to walk you through any part, just let me know—I’m here to help!

review my entrypoint provide currect approach let me know what was different async def entrypoint(ctx: JobContext): """ Inbound call entrypoint that handles incoming SIP calls. Args: ctx: Job context containing call metadata """ logger.info(f"connecting to room {ctx.room.name}") await ctx.connect() try: # Parse call metadata metadata = {} if ctx.job.metadata: metadata = json.loads(ctx.job.metadata) agent_id_raw = metadata.get("agent_id") caller_number = metadata.get("phone_number") or metadata.get("caller_number") if not agent_id_raw: logger.error("No agent_id provided in metadata") raise ValueError("Missing required agent_id in call metadata") # Convert agent_id to integer if it's a string try: agent_id = int(agent_id_raw) except (ValueError, TypeError): logger.error(f"Invalid agent_id format: {agent_id_raw}") raise ValueError(f"Invalid agent_id format: {agent_id_raw}") # Create services knowledge_service, usage_service, agent_service, user_service = ( create_services() ) # Load agent configuration agent_config = await load_agent_config(agent_service, agent_id) user_id = agent_config["user_id"] # Check usage limits if not await check_usage_limits(usage_service, user_service, user_id): logger.warning(f"Usage limit exceeded for user {user_id}") raise ValueError(f"Daily usage limit exceeded for user {user_id}") # Create the inbound agent agent = InboundAgent( agent_config=agent_config, caller_number=caller_number, knowledge_service=knowledge_service, usage_service=usage_service, agent_service=agent_service, user_service=user_service, ) # Create agent session session = create_agent_session() # Start call tracking await agent.start_call_tracking() # Start the session await session.start( agent=agent, room=ctx.room, room_input_options=get_room_input_options(), ) logger.info(f"Inbound agent session started for room {ctx.room.name}") except Exception as e: logger.error(f"Inbound call failed: {e}") # Let the exception propagate - LiveKit will handle cleanup

Your entrypoint implementation looks really well put together and follows best practices for handling SIP calls in LiveKit. I’ve jotted down a few thoughts and suggestions that might help you polish it even more: 1. Metadata Parsing: You’re on the right track with how you’re parsing metadata. Just double-check that the metadata structure stays consistent and includes all the key fields like

agent_id

and

caller_number

. 2. Error Handling: Nice job with the error handling around missing or invalid

agent_id

. This kind of attention really helps with debugging and keeps things running smoothly. 3. Service Initialization: Using

create_services()

and

load_agent_config()

is a smart move for keeping your code modular. Just make sure these functions handle exceptions nicely and log useful error messages. 4. Usage Limits: Including a check for usage limits with

check_usage_limits()

is a solid addition. It might be helpful to log some detailed info if the limit is exceeded to make troubleshooting easier. 5. Agent Session: Creating an

InboundAgent

and kicking off the session with

session.start()

looks great. Just confirm that

get_room_input_options()

is returning the right setup for your specific needs. 6. Logging: Your logging coverage is thorough. Just a heads-up to avoid logging sensitive info like

agent_id

or

caller_number

in plaintext to keep things secure. 7. Graceful Cleanup: The

finally

block doing cleanup with LiveKit is a nice touch. You might also want to explicitly disconnect the agent or free up resources if that fits your workflow. All in all, you’ve built a solid implementation here. If you want me to dive deeper or provide some specific code examples, just let me know—I’m happy to help! Sources: telephony/sip_lifecycle.py | README.md | ivr-nagivator/flask_livekit/extension.py | agent.py | README.md

but this not joining room which is getting created by sip call with disaptch rule

To help your agent join the room created by a SIP call with a dispatch rule, here’s a friendly step-by-step guide: 1. Create a SIP Dispatch Rule: Set up a rule that tells the system which agent should join the room. Here’s an example:

{
  "dispatch_rule": {
    "rule": {
      "dispatchRuleIndividual": {
        "roomPrefix": "sip-call-"
      }
    },
    "roomConfig": {
      "agents": [
        {
          "agentName": "my-agent",
          "metadata": "{\"custom\":\"data\"}"
        }
      ]
    }
  }
}

2. Apply the Dispatch Rule: Use the LiveKit CLI to apply your rule to the SIP trunk like this:

lk sip dispatch create dispatch-rule.json

3. Explicit Agent Dispatch: Make sure your agent is registered with an

agent_name

so you can dispatch them explicitly. For example:

opts = WorkerOptions(agent_name="test-agent")

4. Dispatch via API: You can also use the API to send your agent to the room directly:

dispatch = await lkapi.agent_dispatch.create_dispatch(
    api.CreateAgentDispatchRequest(
        agent_name="test-agent", 
        room="sip-call-room", 
        metadata="{\"user_id\":\"12345\"}"
    )
)

5. Handle Metadata: If you want to pass custom info to your agent, just add it in the

metadata

field—it’ll be included in the job metadata. Following these steps will make sure your agent joins the SIP-created room smoothly. If you have any questions or want me to walk you through anything, just let me know—I’m here to help! Sources: Agent dispatch | LiveKit Docs | packages/livekit-server-sdk/src/AgentDispatchClient.ts

can provide agent.py for dispach rule

{

"dispatch_rule": {

"rule": {

"dispatchRuleIndividual": {

"roomPrefix": "sip-call-"

}

},

"roomConfig": {

"agents": [{

"agentName": "my-agent",

"metadata": "{\"custom\":\"data\"}"

}]

}

}

}

Here’s a friendly example implementation of

agent.py

for your dispatch rule that you might find helpful:

# agent.py
import asyncio
from livekit import api

room_name = "sip-call-room"
agent_name = "my-agent"

async def create_explicit_dispatch():
    lkapi = api.LiveKitAPI()

    # This function creates an explicit dispatch so your agent can join a specific room
    dispatch = await lkapi.agent_dispatch.create_dispatch(
        api.CreateAgentDispatchRequest(
            agent_name=agent_name,
            room=room_name,
            metadata="{\"custom\":\"data\"}"
        )
    )
    print("Created dispatch:", dispatch)

    # Let's also list all dispatches in the room to keep track
    dispatches = await lkapi.agent_dispatch.list_dispatch(room_name=room_name)
    print(f"There are {len(dispatches)} dispatches in {room_name}")
    await lkapi.aclose()

async def create_token_with_agent_dispatch() -> str:
    # Here we create an access token that includes the agent configuration
    token = (
        api.AccessToken()
        .with_identity("my_participant")
        .with_grants(api.VideoGrants(room_join=True, room=room_name))
        .with_room_config(
            api.RoomConfiguration(
                agents=[api.RoomAgentDispatch(agent_name=agent_name, metadata="{\"custom\":\"data\"}")],
            ),
        )
        .to_jwt()
    )
    return token

async def main():
    token = await create_token_with_agent_dispatch()
    print("Created participant token:", token)
    print("Creating explicit dispatch")
    await create_explicit_dispatch()

asyncio.run(main())

Here are a few key points to keep in mind: • Explicit Dispatch: The

create_explicit_dispatch

function helps your agent join the room using the LiveKit API. • Token Configuration: The

create_token_with_agent_dispatch

function generates a token that includes the agent’s settings and metadata. • Room Configuration: The token contains

RoomConfiguration

with

RoomAgentDispatch

to specify the agent and its metadata. Just swap out

room_name

and

agent_name

with your own values. If you have any questions or want me to walk you through anything else, just let me know—I’m here to help!

ahh! this line it should be in docs

*Explicit Dispatch:* Once agents are specified, automatic dispatch takes a backseat. You'll need to assign the agent to a room yourself, either through the API or when generating the token.

more specifically this one ``*Explicit Dispatch:* Once agents are specified, automatic dispatch takes a backseat.` it is not for Worker option even in dispatch rule

You're totally right! When agents are specified, automatic dispatch gets overridden, so you do need to set up explicit dispatch. This isn’t just for WorkerOptions but also for dispatch rules. That means you’ll want to assign agents to rooms directly through the API or when generating tokens. Thanks so much for bringing this up—it’s an important detail that definitely deserves to be clear. If you have any questions or want a hand with your setup, just let me know!

how metadata in dispatch rule get used and by whom

Metadata in a dispatch rule is actually pretty handy and is used mainly by two groups: 1. Agents: When an agent gets dispatched through a dispatch rule, the metadata comes along in the job context. This means the agent can tap into that info to handle tasks better or tweak how it works. For example, it might have user IDs, names, or phone numbers to give some helpful context. 2. Participants: Everyone who joins in via inbound SIP participants created by the dispatch rule also gets this metadata. Apps or systems that work with these participants can then use the metadata for things like filtering, logging, or other handy operations. So, metadata helps keep things consistent across agents and participants, making call handling smoother and more flexible. If you want me to explain anything more, just let me know—I’m happy to help!

update my entrypoint func to join room created by dispatch rule which same agent name along with metadata async def entrypoint(ctx: JobContext): """ Inbound call entrypoint that handles incoming SIP calls. Args: ctx: Job context containing call metadata """ logger.info(f"connecting to room {ctx.room.name}") await ctx.connect() try: # Parse call metadata metadata = {} if ctx.job.metadata: metadata = json.loads(ctx.job.metadata) agent_id_raw = metadata.get("agent_id") caller_number = metadata.get("phone_number") or metadata.get("caller_number") if not agent_id_raw: logger.error("No agent_id provided in metadata") raise ValueError("Missing required agent_id in call metadata") # Convert agent_id to integer if it's a string try: agent_id = int(agent_id_raw) except (ValueError, TypeError): logger.error(f"Invalid agent_id format: {agent_id_raw}") raise ValueError(f"Invalid agent_id format: {agent_id_raw}") # Create services knowledge_service, usage_service, agent_service, user_service = ( await create_services() ) # Load agent configuration agent_config = await load_agent_config(agent_service, agent_id) user_id = agent_config["user_id"] # Check usage limits if not await check_usage_limits(usage_service, user_service, user_id): logger.warning(f"Usage limit exceeded for user {user_id}") raise ValueError(f"Daily usage limit exceeded for user {user_id}") # Create the inbound agent agent = InboundAgent( agent_config=agent_config, caller_number=caller_number, knowledge_service=knowledge_service, usage_service=usage_service, agent_service=agent_service, user_service=user_service, ) # Create agent session session = create_agent_session() # Start call tracking await agent.start_call_tracking() # Start the session await session.start( agent=agent, room=ctx.room, room_input_options=get_room_input_options(), ) logger.info(f"Inbound agent session started for room {ctx.room.name}") except Exception as e: logger.error(f"Inbound call failed: {e}") # Let the exception propagate - LiveKit will handle cleanup

Here’s the updated

entrypoint

function to make sure the agent joins the room created by the dispatch rule with the same agent name and metadata:

async def entrypoint(ctx: JobContext):
    """
    Inbound call entrypoint that handles incoming SIP calls.

    Args:
        ctx: Job context containing call metadata
    """
    <http://logger.info|logger.info>(f"Connecting to room {ctx.room.name}")
    await ctx.connect()

    try:
        # Parse call metadata
        metadata = {}
        if ctx.job.metadata:
            metadata = json.loads(ctx.job.metadata)

        agent_id_raw = metadata.get("agent_id")
        caller_number = metadata.get("phone_number") or metadata.get("caller_number")

        if not agent_id_raw:
            logger.error("No agent_id provided in metadata")
            raise ValueError("Missing required agent_id in call metadata")

        # Convert agent_id to integer if it's a string
        try:
            agent_id = int(agent_id_raw)
        except (ValueError, TypeError):
            logger.error(f"Invalid agent_id format: {agent_id_raw}")
            raise ValueError(f"Invalid agent_id format: {agent_id_raw}")

        # Create services
        knowledge_service, usage_service, agent_service, user_service = (
            await create_services()
        )

        # Load agent configuration
        agent_config = await load_agent_config(agent_service, agent_id)
        user_id = agent_config["user_id"]

        # Check usage limits
        if not await check_usage_limits(usage_service, user_service, user_id):
            logger.warning(f"Usage limit exceeded for user {user_id}")
            raise ValueError(f"Daily usage limit exceeded for user {user_id}")

        # Make sure the agent joins the room created by the dispatch rule
        if ctx.room.name.startswith("sip-call-") and agent_config["agent_name"] == ctx.job.metadata.get("agent_name"):
            <http://logger.info|logger.info>(f"Agent {agent_config['agent_name']} is joining the room {ctx.room.name}")

        # Create the inbound agent
        agent = InboundAgent(
            agent_config=agent_config,
            caller_number=caller_number,
            knowledge_service=knowledge_service,
            usage_service=usage_service,
            agent_service=agent_service,
            user_service=user_service,
        )

        # Create agent session
        session = create_agent_session()

        # Start call tracking
        await agent.start_call_tracking()

        # Start the session
        await session.start(
            agent=agent,
            room=ctx.room,
            room_input_options=get_room_input_options(),
        )

        <http://logger.info|logger.info>(f"Inbound agent session started for room {ctx.room.name}")

    except Exception as e:
        logger.error(f"Inbound call failed: {e}")
        # Let the exception propagate - LiveKit will handle cleanup

Here are the key updates: • We added a check to make sure the agent joins the right room by verifying the room name starts with

sip-call-

and that the agent name in the metadata matches. • Included some logging so you can see when the agent is joining the correct room. If you want me to tweak anything else or have questions, just let me know—I’m happy to help!

this dispatch rule { "sipDispatchRuleId": "SDR_xxxxxxx", "rule": { "dispatchRuleIndividual": { "roomPrefix": "call-" } }, "trunkIds": [ "testtrunck id" ], "name": "voice-ai-universal-agent dispatch rule", "roomConfig": { "agents": [ { "agentName": "voice-ai-universal-agent", "metadata": "{\"agent_id\": 60, \"agent_external_id\": \"some id\", \"agent_type\": \"inbound\"}" } ] } }

Thanks so much for sharing the dispatch rule! Here’s a quick rundown based on what you’ve provided: • Room Prefix: Any room created by this rule will have names starting with

call-

. • Agent: The agent

voice-ai-universal-agent

will be the one dispatched to these rooms. • Metadata: The agent will get the following metadata:

{
  "agent_id": 60,
  "agent_external_id": "some id",
  "agent_type": "inbound"
}

• Trunk: This rule is linked to the SIP trunk ID

testtrunck id

. Just make sure your entrypoint function looks out for the room prefix

call-

and matches the agent name

voice-ai-universal-agent

in the metadata so it can join the room properly. If you have any questions or want to chat more about this, I’m here to help!

where this can be check call-

Hey there! You can check if the room prefix is

call-

by looking at the

ctx.room.name

property in your entrypoint function. Here’s a quick example:

if ctx.room.name.startswith("call-"):
    <http://logger.info|logger.info>(f"Room {ctx.room.name} matches the dispatch rule prefix.")

This way, your agent will only handle rooms created by the dispatch rule with that prefix. If anything’s unclear or you want to dive deeper, just give me a shout—I’m happy to help!

can we have multiple dispatch rule with different metdata and diffrent trunck id

Absolutely! LiveKit lets you set up multiple dispatch rules, each with its own metadata and trunk IDs. Here’s what each dispatch rule can do: • Link to specific trunk IDs. • Include unique metadata to tailor things just right. • Have different settings like room prefixes, pins, and attributes. For instance, you might create one dispatch rule for

trunk_id_1

with metadata

{"agent_id": 1}

, and another for

trunk_id_2

with metadata

{"agent_id": 2}

. The system will then check the rules and apply the best match based on trunk ID and setup. If you want, I’m happy to help you get those multiple dispatch rules up and running! Sources: SIP overview | LiveKit Docs | SIP overview | LiveKit Docs | SIP APIs | LiveKit Docs | livekit/livekit_sip.pb.go

what is limit on dispatch rule and truck

Great news! LiveKit doesn’t set hard limits on how many dispatch rules or trunks you can create. That said, there are a few important things to keep in mind when setting things up: • Room Prefix Sharing: You can have multiple dispatch rules using the same room prefix, but they need to be uniquely identified by things like PIN, trunk ID, or inbound number to keep everything running smoothly. • Conflict Prevention: The system checks each rule based on a mix of PIN, trunk ID, and inbound number, so it won’t allow duplicates with the exact same combination. • Trunk Association: Feel free to link multiple dispatch rules to the same trunk, just make sure each has its own unique PIN. • Priority System: If a call matches several rules, LiveKit sorts them out by prioritizing PINs, specific numbers, and rules tied to trunks. If you’d like me to walk you through any of this or help with setting up your dispatch rules and trunks, just let me know—I’m here to help! Sources: SIP overview | LiveKit Docs | livekit/livekit_sip.pb.go

SO i have a common agent which will handle call based metdata comming from differet trunck and dispatch rule

To handle calls with a common agent across different trunks and dispatch rules, here’s a friendly approach you can try out: 1. Metadata-Driven Behavior: Make sure each dispatch rule includes metadata that’s specific to the trunk or rule. This way, the common agent can use that info to figure out the best way to handle each call. 2. Entrypoint Logic: In the agent’s entrypoint function, take a look at the metadata from the job context (

ctx.job.metadata

) to spot the trunk, dispatch rule, or any other important details. For example:

metadata = json.loads(ctx.job.metadata)
trunk_id = metadata.get("trunk_id")
dispatch_rule_id = metadata.get("dispatch_rule_id")
# Use these values to customize agent behavior

3. Dynamic Configuration: Use that metadata to load configurations on the fly, like call scripts, routing logic, or service integrations, depending on the trunk or dispatch rule. 4. Unified Agent: Register one single agent (like

common-agent

) across all dispatch rules. This agent will handle every call and adjust how it behaves based on the metadata it receives. This setup lets one agent handle calls smoothly and flexibly across multiple trunks and dispatch rules. If you’d like, I’m here to help you get this up and running!

give sample code for same agents

Here’s a sample implementation for a common agent handling calls based on metadata from different trunks and dispatch rules:

import json

async def entrypoint(ctx: JobContext):
    """
    Common agent entrypoint to handle calls based on metadata.

    Args:
        ctx: Job context containing call metadata
    """
    <http://logger.info|logger.info>(f"Connecting to room {ctx.room.name}")
    await ctx.connect()

    try:
        # Parse metadata from the job context
        metadata = json.loads(ctx.job.metadata) if ctx.job.metadata else {}

        # Extract relevant metadata fields
        trunk_id = metadata.get("trunk_id")
        dispatch_rule_id = metadata.get("dispatch_rule_id")
        caller_number = metadata.get("phone_number")

        if not trunk_id or not dispatch_rule_id:
            logger.error("Missing trunk_id or dispatch_rule_id in metadata")
            raise ValueError("Required metadata fields are missing")

        # Log metadata for debugging
        <http://logger.info|logger.info>(f"Handling call from trunk: {trunk_id}, dispatch rule: {dispatch_rule_id}")

        # Dynamically load configurations or behavior based on metadata
        if dispatch_rule_id == "rule_1":
            <http://logger.info|logger.info>("Applying behavior for dispatch rule 1")
            # Add specific logic for rule 1
        elif dispatch_rule_id == "rule_2":
            <http://logger.info|logger.info>("Applying behavior for dispatch rule 2")
            # Add specific logic for rule 2
        else:
            <http://logger.info|logger.info>("Applying default behavior")

        # Create the agent and session
        agent = CommonAgent(caller_number=caller_number, metadata=metadata)
        session = create_agent_session()

        # Start the session
        await session.start(
            agent=agent,
            room=ctx.room,
            room_input_options=get_room_input_options(),
        )

        <http://logger.info|logger.info>(f"Agent session started for room {ctx.room.name}")

    except Exception as e:
        logger.error(f"Call handling failed: {e}")
        # Let the exception propagate for cleanup

Key Features: • Metadata Parsing: It gently pulls out

trunk_id

,

dispatch_rule_id

, and other handy details from the job metadata. • Dynamic Behavior: The agent adapts its approach depending on the dispatch rule or trunk ID—pretty neat, right? • Unified Agent: We use one friendly agent instance (

CommonAgent

) to handle all calls, keeping things simple and smooth. If you’d like me to tweak this or need some more examples, just let me know—I’m here to help!

I am running from here i have shared entrypoint earlier """ Main Agent Runner - LiveKit Agents SDK Implementation Main script to run voice AI agents using the LiveKit agents framework. Supports both inbound and outbound SIP calling with knowledge base integration. """ import json import logging import os from livekit.agents import cli, WorkerOptions, JobContext, JobRequest from dotenv import load_dotenv # Load environment variables from .env file load_dotenv() from pkg.config import config # Setup logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger("voice-ai-agents") def validate_environment(): """ Validate required environment variables. Raises: ValueError: If required environment variables are missing """ required_vars = [ "AGENT_DB_URL", "LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET", ] missing_vars = [] for var in required_vars: if not getattr(config, var, None): missing_vars.append(var) if missing_vars: raise ValueError(f"Missing required environment variables: {missing_vars}") def determine_agent_type(ctx: JobContext) -> str: """ Determine agent type from job context metadata. Args: ctx: Job context containing metadata Returns: Agent type ('inbound' or 'outbound') """ try: logger.info("determining agent type") if ctx.job.metadata: metadata = json.loads(ctx.job.metadata) agent_type = metadata.get("agent_type", "inbound") # If phone_number is provided but no explicit type, assume outbound if not metadata.get("agent_type") and metadata.get("phone_number"): agent_type = "outbound" return agent_type except (json.JSONDecodeError, KeyError) as e: logger.warning(f"Could not parse job metadata: {e}") # Default to inbound return "inbound" async def enhanced_request_handler(ctx: JobRequest) -> None: """ Enhanced job request handler with intelligent filtering. This function decides whether to accept incoming jobs based on: - Agent availability and load - Job metadata and requirements - User plan limits and usage """ try: # Log the job request logger.info(f"Job request received for room: {ctx.job.room.name}") # Parse metadata for agent requirements metadata = {} if hasattr(ctx.job, 'metadata') and ctx.job.metadata: try: metadata = json.loads(ctx.job.metadata) except json.JSONDecodeError: logger.warning("Failed to parse job metadata") # Extract agent_id for validation agent_id = metadata.get("agent_id") if not agent_id: logger.warning("Job request missing agent_id, rejecting") await ctx.reject() return # Quick validation - could add more sophisticated checks here: # - User plan validation # - Agent configuration validation # - Resource availability checks logger.info(f"Accepting job for agent_id: {agent_id}") await ctx.accept( identity=f"agent-{agent_id}", name="Voice AI Agent", metadata=json.dumps({"agent_id": agent_id, "worker_type": "universal"}) ) except Exception as e: logger.error(f"Error in job request handler: {e}") await ctx.reject() def calculate_worker_load() -> float: """ Calculate current worker load based on multiple factors. Returns: Float between 0.0 (idle) and 1.0 (fully loaded) """ try: import psutil # Get CPU usage cpu_percent = psutil.cpu_percent(interval=0.1) # Get memory usage memory = psutil.virtual_memory() memory_percent = memory.percent # Calculate combined load (weighted average) combined_load = (cpu_percent * 0.6 + memory_percent * 0.4) / 100.0 logger.debug(f"Worker load - CPU: {cpu_percent}%, Memory: {memory_percent}%, Combined: {combined_load:.2f}") return min(combined_load, 1.0) except ImportError: # Fallback if psutil not available logger.warning("psutil not available, using basic load calculation") return 0.1 # Conservative default except Exception as e: logger.error(f"Error calculating load: {e}") return 0.5 # Safe middle value def prewarm_agent_process(proc) -> None: """ Prewarm agent process for faster job startup. This function runs when a new process starts, before any jobs are assigned. Use it to load models, initialize connections, or warm up resources. """ import os import sys # Add project root to path project_root = os.path.dirname(os.path.abspath(file)) if project_root not in sys.path: sys.path.insert(0, project_root) try: logger.info(f"Prewarming agent process {proc.pid if hasattr(proc, 'pid') else 'unknown'}") # Preload common imports to reduce startup time import json import asyncio from livekit.agents import AgentSession from livekit.plugins import openai, deepgram, cartesia # Preload our agent modules from agents.common import create_services, configure_voice_pipeline from agents.inbound_agent import InboundAgent from agents.outbound_agent import OutboundAgent # Pre-configure voice pipeline to cache models try: configure_voice_pipeline() logger.info("Voice pipeline pre-configured successfully") except Exception as e: logger.warning(f"Voice pipeline pre-configuration failed: {e}") # Pre-initialize database connection pool for faster access try: import asyncio loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) async def init_db(): from db.postgres.connection_manager import DBConnectionManager await DBConnectionManager.initialize_pool(os.getenv('AGENT_DB_URL')) loop.run_until_complete(init_db()) loop.close() logger.info("Database connection pool pre-initialized") except Exception as e: logger.warning(f"Database pre-initialization failed: {e}") logger.info("Agent process prewarm completed successfully") except Exception as e: logger.error(f"Agent process prewarm failed: {e}") async def agent_entrypoint(ctx: JobContext): """ Main agent entrypoint that routes to appropriate agent type. Args: ctx: Job context containing room and metadata information """ try: # Determine agent type from metadata agent_type = determine_agent_type(ctx) logger.info(f"Routing to {agent_type} agent for room {ctx.room.name}") # Route to appropriate agent entrypoint if agent_type == "outbound": from agents.outbound_agent import entrypoint as outbound_entrypoint await outbound_entrypoint(ctx) else: from agents.inbound_agent import entrypoint as inbound_entrypoint await inbound_entrypoint(ctx) except Exception as e: logger.error(f"Failed to create agent: {e}") # Let the exception propagate - LiveKit will handle cleanup raise def run_universal_agent(): """Run universal agent that handles both inbound and outbound calls.""" logger.info("Starting Universal Voice Agent (handles both inbound and outbound)") worker_options = WorkerOptions( entrypoint_fnc=agent_entrypoint, agent_name="voice-ai-universal-agent", # Explicit dispatch - requires dispatch rules request_fnc=enhanced_request_handler, # Custom job filtering prewarm_fnc=prewarm_agent_process, # Prewarm processes for faster startup # load_fnc=calculate_worker_load, # Custom load calculation load_threshold=0.8, # Accept jobs until 80% load num_idle_processes=2, # Keep 2 warm processes ready ) cli.run_app(worker_options) def main(): """ Main function to run the voice AI agent. """ try: # Validate environment validate_environment() # Run universal agent that handles both inbound and outbound run_universal_agent() except KeyboardInterrupt: logger.info("Agent shutdown requested") except Exception as e: logger.error(f"Agent failed: {e}") raise if name == "__main__": main()

Thanks so much for sharing the detailed implementation! From what I see, your universal agent is nicely set up to handle both inbound and outbound calls smoothly. Here are a few key points and some friendly suggestions: 1. Agent Name: You've named the agent

voice-ai-universal-agent

, which matches your dispatch rule perfectly. Just double-check that all dispatch rules pointing to this name are set up correctly. 2. Metadata Handling: Your

determine_agent_type

and

enhanced_request_handler

functions do a great job parsing metadata to pick the right agent type and validate job requests. This really helps keep things flexible. 3. Entrypoint Routing: The

agent_entrypoint

function smartly routes calls to either the inbound or outbound agent based on metadata. It’s a neat, modular way to keep things organized and scalable. 4. Prewarming: The

prewarm_agent_process

function loads up resources like the voice pipeline and database connections ahead of time, which helps cut down on startup delays. Just make sure these preloaded parts are used often enough to make the extra prep worthwhile. 5. Load Management: I really like how the

calculate_worker_load

function keeps track of worker load dynamically. It’s a smart move to balance resources and keep performance running smoothly. If you’d like a hand with any specific parts or want some more examples, just give me a shout—I’m here to help!

i am making call to twilio number i am getting ring continuesly, i can see room is getting created on livekit cloud for all but agent is not joining the room

Let’s walk through some steps to figure out why the agent isn’t joining the room created by the SIP call: 1. SIP Call Status: First, check that the SIP call actually connects. You can look at the

sip.callStatus

in the participant’s metadata—it should say "active" once the call is connected. 2. Dispatch Rules: Make sure the SIP dispatch rule is pointing to the right agent in

room_config.agents

. Double-check that the agent name matches exactly what’s registered with LiveKit. 3. Agent Availability: Confirm the agent is properly registered and running. The agent needs to be active to receive any dispatch requests. 4. Logs: Take a peek at the SIP service logs for any errors during call setup or agent dispatch. Watch out for issues like incorrect metadata or dispatch failures. 5. Room Existence: Verify that the room actually gets created when the SIP call connects. The agent can only join a room that exists. 6. Token Configuration: If you’re using token-based dispatch, make sure the token includes the right

RoomAgentDispatch

configuration. If you want, I’m here to help dig into specific logs or configurations to get this sorted out! Sources: Dispatching agents | LiveKit Docs | pkg/sip/room.go | Web and mobile frontends | LiveKit Docs | protobufs/livekit_sip.proto | test/agent_test.go

i have shared dispatch rule earlier how can enabled debug mode to check where it is failing silenty i am not see any new log in agent process

If you want to enable debug mode for LiveKit agents and get to the bottom of those silent failures, here are a few easy ways to do it: 1. Set Log Level: Just use the

setLogLevel

function to turn on detailed logging:

import { setLogLevel, LogLevel } from 'livekit-client';
setLogLevel(LogLevel.debug);

2. Environment Variable: You can set

LIVEKIT_EVALS_VERBOSE=1

to see more verbose output:

LIVEKIT_EVALS_VERBOSE=1 pytest -s -o log_cli=true <your-test-file>

3. Run in Dev Mode: Start the agent in development mode—it’s super helpful for debugging:

python agent.py dev

4. Console Mode: Running the agent in console mode lets you debug locally:

python agent.py console

These tips should help you grab detailed logs and figure out what’s going on. If you hit any snags or want a hand, just let me know—I’m here to help! Sources: Testing and evaluation | LiveKit Docs | Voice AI quickstart | LiveKit Docs | hooks/useDebug.ts | avatars/hedra/education_avatar/education-frontend/hooks/useDebug.ts

is there way to test dispatch to room directly from livekit ?

Hey there! You can easily test agent dispatch to a room directly in LiveKit using any of these methods: 1. Using the Python SDK:

import asyncio
from livekit import api

room_name = "my-room"
agent_name = "test-agent"

async def create_explicit_dispatch():
    lkapi = api.LiveKitAPI()
    dispatch = await lkapi.agent_dispatch.create_dispatch(
        api.CreateAgentDispatchRequest(
            agent_name=agent_name, room=room_name, metadata="my_job_metadata"
        )
    )
    print("Dispatch created:", dispatch)

    dispatches = await lkapi.agent_dispatch.list_dispatch(room_name=room_name)
    print(f"Found {len(dispatches)} dispatches in {room_name}")

asyncio.run(create_explicit_dispatch())

2. Using the Node.js SDK: `const { AgentDispatchClient } = require('livekit-server-sdk'); const roomName = 'my-room'; const agentName = 'test-agent'; const client = new AgentDispatchClient('LIVEKIT_URL', 'API_KEY', 'API_SECRET'); const dispatch = await client.createDispatch(roomName, agentName, { metadata: '{"mydata": "myvalue"}', }); console.log('Dispatch created:', dispatch); const dispatches = await client.listDispatch(roomName); console.log(

Found ${dispatches.length} dispatches in ${roomName}

);` 3. Using the LiveKit CLI:

lk dispatch create --room test-room --agent-name test-agent

These are great ways to explicitly dispatch an agent to a room for testing. If you have any questions or want a hand with anything else, just let me know—I’m here to help! Sources: packages/livekit-server-sdk/src/AgentDispatchClient.ts | Dispatching agents | LiveKit Docs | Agent dispatch | LiveKit Docs

2025-08-07 171257,224 - voice-ai-agents - INFO - Job request received for room: test-room 2025-08-07 171257,224 - INFO voice-ai-agents - Job request received for room: test-room 2025-08-07 171257,224 - voice-ai-agents - WARNING - Job request missing agent_id, rejecting 2025-08-07 171257,224 - WARNING voice-ai-agents - Job request missing agent_id, rejecting which is okay these logs which expecting some medata

Hey there! The logs show that the job request is getting rejected because the

agent_id

is missing from the metadata. This makes sense given how your current setup works—it checks for

agent_id

before accepting any job. To fix this, just make sure the dispatch rule or token you're using to create the job includes the

agent_id

in the metadata. Here’s a quick example:

{
  "agents": [
    {
      "agentName": "voice-ai-universal-agent",
      "metadata": "{\"agent_id\": 60, \"agent_external_id\": \"some id\", \"agent_type\": \"inbound\"}"
    }
  ]
}

If you want, I’m here to help you update the dispatch rule or token configuration—just let me know!

give more example of explicit dispatch

Here are some extra examples showing how to explicitly dispatch using different SDKs—hope these help! Python SDK:

import asyncio
from livekit import api

async def create_explicit_dispatch():
    lkapi = api.LiveKitAPI()

    # Dispatch an agent to a specific room with metadata
    dispatch = await lkapi.agent_dispatch.create_dispatch(
        api.CreateAgentDispatchRequest(
            agent_name="test-agent",
            room="test-room",
            metadata="{\"key\": \"value\"}"
        )
    )
    print("Dispatch created:", dispatch)

    # List all dispatches in the room
    dispatches = await lkapi.agent_dispatch.list_dispatch(room_name="test-room")
    print(f"Found {len(dispatches)} dispatches in test-room")

asyncio.run(create_explicit_dispatch())

Node.js SDK:

const { AgentDispatchClient } = require('livekit-server-sdk');

async function createExplicitDispatch() {
  const client = new AgentDispatchClient('LIVEKIT_URL', 'API_KEY', 'API_SECRET');

  // Dispatch an agent to a specific room with metadata
  const dispatch = await client.createDispatch('test-room', 'test-agent', {
    metadata: '{"key": "value"}'
  });
  console.log('Dispatch created:', dispatch);

  // List all dispatches in the room
  const dispatches = await client.listDispatch('test-room');
  console.log(`Found ${dispatches.length} dispatches in test-room`);
}

createExplicitDispatch();

LiveKit CLI:

# Create a dispatch using the CLI
lk dispatch create --room test-room --agent-name test-agent --metadata '{"key": "value"}'

# List all dispatches in a room
lk dispatch list --room test-room

These examples show you how to explicitly dispatch agents to a room with metadata using Python, Node.js, and the LiveKit CLI. If you have any questions or want to dive deeper, just let me know—I'm here to help!

give python code one where i can put this code and what i should listen for to dispatch agent for inbound call from sip

Here’s the Python code for explicit dispatch:

import asyncio
from livekit import api

async def create_explicit_dispatch():
    lkapi = api.LiveKitAPI()

    # Dispatch an agent to a specific room with metadata
    dispatch = await lkapi.agent_dispatch.create_dispatch(
        api.CreateAgentDispatchRequest(
            agent_name="voice-ai-universal-agent",
            room="sip-call-room",
            metadata="{\"agent_id\": 60, \"agent_type\": \"inbound\"}"
        )
    )
    print("Dispatch created:", dispatch)

    # List all dispatches in the room
    dispatches = await lkapi.agent_dispatch.list_dispatch(room_name="sip-call-room")
    print(f"Found {len(dispatches)} dispatches in sip-call-room")

asyncio.run(create_explicit_dispatch())

Where to Place This Code: • Feel free to add this code in your agent’s initialization or a special script just for testing out dispatch functionality. • When you’re ready for production, you can tuck it into your SIP call handling workflow—like right when a SIP call comes in and a room gets created. What to Listen For: • Keep an ear out for SIP call events, such as when a call connects or a room is created. You can use the

JobContext

or

JobRequest

in your agent’s entrypoint to manage these events. • Check

ctx.room.name

and

ctx.job.metadata

to decide if it’s the right time to dispatch the agent. If you want a hand integrating this into your workflow, just let me know—I’m happy to help!

i want to dispatch agent when room get created by dispatch rule from sip call

Here’s a friendly guide to help you dispatch an agent whenever a room is created by a dispatch rule from a SIP call: 1. Listen for Room Creation: In your agent’s entrypoint, keep an eye out using

JobRequest

or

JobContext

to spot when a room pops up. For example:

async def enhanced_request_handler(ctx: JobRequest):
    if ctx.job.room.name.startswith("sip-call-"):
        <http://logger.info|logger.info>(f"Room created: {ctx.job.room.name}")
        await dispatch_agent(ctx.job.room.name)

2. Dispatch the Agent: Then, you can explicitly assign the agent to that room with this code:

async def dispatch_agent(room_name):
    lkapi = api.LiveKitAPI()
    dispatch = await lkapi.agent_dispatch.create_dispatch(
        api.CreateAgentDispatchRequest(
            agent_name="voice-ai-universal-agent",
            room=room_name,
            metadata="{\"agent_id\": 60, \"agent_type\": \"inbound\"}"
        )
    )
    <http://logger.info|logger.info>(f"Agent dispatched to room: {room_name}")

3. Integrate with SIP Workflow: Make sure your SIP dispatch rule creates rooms with a clear prefix like

sip-call-

. This way, your handler can easily spot and manage them. This setup will automatically dispatch an agent whenever a SIP call creates a room. If you have any questions or want to chat more about it, just let me know—I’m here to help!

when i create explicit dispatch where i can get disparule metadata which essential for multi tenant system

When you create an explicit dispatch, the metadata defined in the dispatch rule is passed to the agent as part of the

JobContext

. You can easily access it in the agent’s entrypoint function using

ctx.job.metadata

. Here’s a quick example:

async def entrypoint(ctx: JobContext):
    # Parse metadata from the job context
    metadata = json.loads(ctx.job.metadata) if ctx.job.metadata else {}

    # Access dispatch rule metadata
    tenant_id = metadata.get("tenant_id")
    agent_id = metadata.get("agent_id")

    <http://logger.info|logger.info>(f"Dispatch rule metadata: Tenant ID: {tenant_id}, Agent ID: {agent_id}")

This metadata is super important for multi-tenant systems because it helps you identify the tenant and tailor the agent’s behavior just right. Make sure the dispatch rule includes the necessary metadata when you create it. For example:

{
  "agents": [
    {
      "agentName": "voice-ai-universal-agent",
      "metadata": "{\"tenant_id\": \"12345\", \"agent_id\": 60}"
    }
  ]
}

If anything’s unclear or you want to dive deeper, just let me know—I’m here to help!

but when my entrypoint will be executed?

Hey there! Just wanted to explain how the entrypoint function in your agent works when a job gets dispatched to it. Here’s when it kicks in: 1. Explicit Dispatch: When you manually dispatch an agent to a room using the API or CLI, the entrypoint function will run for that agent. 2. Room Creation by Dispatch Rule: If a SIP call triggers a dispatch rule that assigns an agent to a room, the entrypoint function will automatically run for that assigned agent. 3. Job Assignment: The entrypoint runs when the LiveKit server assigns a job to the agent based on your dispatch setup. For your setup, just double-check that your dispatch rule or explicit dispatch is properly assigning the agent to the room. Once the job is assigned, the entrypoint function will be called with the

JobContext

, so you can easily access metadata and handle the job. If anything’s unclear or you want to chat more about it, just let me know—I’m here to help!

but you are asking me to listen for job context in entryopoint and do explict dispatch don't you see probelm hee?

You're totally right—there’s a bit of a logical hiccup in trying to listen for a job context in the entrypoint and then doing an explicit dispatch. The entrypoint actually runs only after a job has already been dispatched to the agent, so it can’t be used to kick off the dispatch itself. Here’s how you can fix this: 1. Use Dispatch Rules: Make sure your SIP dispatch rules are set up to automatically assign the agent to the room as soon as it’s created. This way, you won’t need to do an explicit dispatch after the room is up. 2. Explicit Dispatch Before Job Assignment: If you do want to use explicit dispatch, it needs to be triggered from outside (like a separate service or script) right when the room is created by the SIP call. 3. Room Monitoring: Consider using a monitoring service or webhook to watch for when a room is created and then trigger the explicit dispatch from there. Feel free to reach out if you want a hand setting up dispatch rules or creating an external trigger for explicit dispatch—I’m here to help!

ido agent get metdata set in trunck while join the room?

Hey there! So, an agent can get metadata when joining a room, but it really depends on how that metadata is set up: 1. Explicit Dispatch: If an agent is explicitly dispatched to a room, you can pass metadata directly using the

metadata

field in the

CreateAgentDispatchRequest

. The agent will then receive this metadata in the

JobContext

. 2. SIP Participant Metadata: Metadata set in the SIP trunk (like with

participant_metadata

or

participant_attributes

) gets attached to the SIP participant. However, this metadata won’t automatically reach the agent unless you include it explicitly in the dispatch. So, to make sure your agent gets the metadata they need, it’s best to use explicit dispatch with the metadata field. If you want me to walk you through it or share some examples, just let me know—I’m here to help! Sources: Agent dispatch | LiveKit Docs | livekit/livekit_sip.pb.go

I am thinking updated dispatch rule and moving metadata some where else like in attribute? or trunk dispatch?

You're absolutely right—metadata can be set at different levels, like the trunk or the dispatch rule. Here’s a friendly rundown on how you might want to approach this: 1. Trunk-Level Metadata: When you set metadata at the trunk level, it applies to all calls coming through that trunk. This is great for settings or attributes that stay the same across all calls for a particular trunk. 2. Dispatch Rule-Level Metadata: Metadata at the dispatch rule level is a bit more tailored—it only applies to calls that match that specific rule. This works well for tenant-specific or call-specific setups. Here’s a little tip: If you’re working in a multi-tenant system, you might want to: • Use trunk-level metadata for global tenant identifiers or settings that everyone shares. • Use dispatch rule-level metadata for configurations or attributes that change depending on the tenant or the call. If you’d like, I’m here to help you update your dispatch rules or figure out the best spot for your metadata—just let me know!

{ "sipDispatchRuleId": "SDR_xxxxxxxx", "rule": { "dispatchRuleIndividual": { "roomPrefix": "call-643aaa88-" } }, "trunkIds": [ "ST_Axxxxx" ], "name": "Agent dispatch rule" } is this valid dispatch rule

Hey there! The JSON you shared is a valid LiveKit SIP dispatch rule. Here's a quick rundown: • sipDispatchRuleId: This is the unique ID for the dispatch rule. • rule: It’s set as a

dispatchRuleIndividual

type with a

roomPrefix

of

call-643aaa88-

. • trunkIds: These are the trunk IDs (

ST_Axxxxx

) that this rule applies to. • name: A friendly name for the rule (

Agent dispatch rule

). Everything fits perfectly with LiveKit’s SIP Dispatch Rule requirements. If you have any questions or need a hand with anything else, just let me know—I’m here to help!