can we do multiple dispatches in livekit using python sdk?

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

as of now i am doing dispatch as below:

await asyncio.wait_for(
                self.lkapi.agent_dispatch.create_dispatch(
                    CreateAgentDispatchRequest(
                        agent_name="outbound-call-agent",
                        room=room_name,
                        metadata=json.dumps(metadata)
                    )
                ),
                timeout=10.0
            )

Thanks — that snippet is perfect. To dispatch multiple agents you can call

create_dispatch

multiple times (concurrently or sequentially). Example patterns: 1) Sequential (simple):

for agent in ["agent-a", "agent-b", "agent-c"]:
    await asyncio.wait_for(
        self.lkapi.agent_dispatch.create_dispatch(
            CreateAgentDispatchRequest(
                agent_name=agent,
                room=room_name,
                metadata=json.dumps(metadata)
            )
        ),
        timeout=10.0
    )

2) Concurrent (faster):

tasks = [
    asyncio.create_task(
        self.lkapi.agent_dispatch.create_dispatch(
            CreateAgentDispatchRequest(agent_name=agent, room=room_name, metadata=json.dumps(metadata))
        )
    )
    for agent in ["agent-a", "agent-b", "agent-c"]
]
await asyncio.wait_for(asyncio.gather(*tasks), timeout=10.0)

Choose sequential if you need ordered creation or per-dispatch error handling; choose concurrent to speed up multiple dispatches. Do you want an example that includes per-dispatch retries or individual timeouts? , with interactive elements

but i have only i container with the agent name: outbound-call-agent

Thanks — regarding your setup with a single container named `outbound-call-agent`: • You can create multiple dispatches that target the same agent name, but each active agent instance must be running to accept a dispatch. A single container instance can only handle the number of concurrent dispatches it’s programmed to manage (often one logical agent connection), so dispatches sent while that agent isn’t available will queue or fail depending on your logic. • If you need true parallel agents, run multiple containers/instances using the same agent image but with distinct agent identities (e.g.,

outbound-call-agent-1

,

outbound-call-agent-2

) so each dispatch is delivered to an available instance. • If you want the single container to handle multiple dispatches concurrently, ensure the agent process supports multiple simultaneous connections and accepts multiple dispatches; then dispatching the same

agent_name

to multiple rooms can work (subject to your agent’s concurrency logic). Which direction do you want to pursue? • Scale horizontally (run multiple containers with different agent names) — I can show container/instance naming and dispatch patterns. • Keep a single container and add concurrency support in the agent process — I can outline code patterns for handling multiple simultaneous dispatches. • Or do you want an example of dispatching the same agent name to multiple rooms?, with interactive elements

an example of dispatching the same agent name to multiple rooms

Thanks — regarding dispatching the same agent name to multiple rooms, here’s a concise example and notes. Example: dispatch the same

outbound-call-agent

to multiple rooms concurrently

import asyncio
import json
from livekit import api
from livekit.api import CreateAgentDispatchRequest

async def dispatch_to_rooms(lkapi: api.LiveKitAPI, agent_name: str, rooms: list, metadata: dict):
    tasks = []
    for room in rooms:
        req = CreateAgentDispatchRequest(
            agent_name=agent_name,
            room=room,
            metadata=json.dumps(metadata)
        )
        tasks.append(lkapi.agent_dispatch.create_dispatch(req))

    # run concurrently and await all results
    results = await asyncio.gather(*tasks, return_exceptions=True)
    # results contains responses or exceptions for each dispatch
    return results

# usage
async def main():
    lkapi = api.LiveKitAPI()  # your instantiated client
    rooms = ["room-a", "room-b", "room-c"]
    metadata = {"user_id": "12345"}
    results = await dispatch_to_rooms(lkapi, "outbound-call-agent", rooms, metadata)
    for room, res in zip(rooms, results):
        if isinstance(res, Exception):
            print(f"dispatch to {room} failed: {res}")
        else:
            print(f"dispatched to {room}: {res}")

asyncio.run(main())

Important notes: • Sending dispatches to multiple rooms with the same

agent_name

is allowed. • Whether those dispatches are actually handled concurrently depends on your agent instance: ◦ If your single container supports multiple simultaneous agent connections and logic to handle them, it can accept multiple dispatches. ◦ If it only accepts one active connection, only one dispatch will be served at a time — others may queue or fail. • If you need true parallel handling, run multiple agent instances (distinct identities or same name if your system allows multiple connections for that name) or make the agent process concurrent-capable. Want a version that: • uses per-dispatch timeouts/retries, or • checks dispatch results and retries failed ones?, with interactive elements

what about: Keep a single container and add concurrency support in the agent process

Great — regarding keeping a single container and adding concurrency support in the agent process, here’s a concise plan and a minimal example to get you started. Key approaches (pick one or combine): • Use an async event loop (asyncio) so the agent can maintain multiple LiveKit connections or handle multiple rooms concurrently. • Use worker threads or a thread pool for blocking work (e.g., sync SDK calls or heavy processing) while the main loop handles I/O. • Use a task queue (in-memory with asyncio.Queue or external like Redis/RQ) to schedule and limit concurrent dispatch handling. • Control concurrency limits with semaphores to avoid resource overload (e.g., max concurrent sessions). Minimal asyncio pattern (conceptual): 1. Accept dispatch events (e.g., via webhook, long-poll, or an AgentDispatch consumer). 2. For each dispatch, create an asyncio task to handle the room connection and session lifecycle. 3. Use an asyncio.Semaphore to limit how many session tasks run at once. 4. Use proper cancellation, timeouts, and cleanup to avoid leaked connections. Example (asyncio + semaphore + per-dispatch handler):

import asyncio
import json
from livekit import Room, RoomOptions  # pseudocode imports

MAX_CONCURRENT = 5
semaphore = asyncio.Semaphore(MAX_CONCURRENT)

async def handle_dispatch(dispatch):
    # dispatch contains agent_name, room, metadata, etc.
    async with semaphore:
        try:
            # connect to LiveKit room (pseudo-code; use your SDK's connect)
            room = await Room.connect(dispatch.room, token=dispatch.token, options=RoomOptions())
            # perform agent work: publish/subscribe, handle tracks, signaling, SIP bridging, etc.
            await run_agent_session(room, dispatch.metadata)
        except Exception as e:
            # handle/log error
            raise
        finally:
            # ensure cleanup
            await room.disconnect()

async def dispatch_consumer(dispatch_source):
    # dispatch_source yields incoming dispatch objects
    async for dispatch in dispatch_source():
        asyncio.create_task(handle_dispatch(dispatch))

# Example run
asyncio.run(dispatch_consumer(my_dispatch_source))

Notes: • Replace pseudo-code with the concrete LiveKit Python client calls you use to join rooms and manage tracks. • If the SDK has blocking calls, run them in an executor:

await loop.run_in_executor(...)

. • Add per-dispatch timeouts (asyncio.wait_for) and retries as needed. • Monitor resource usage (CPU, memory, sockets) and tune MAX_CONCURRENT accordingly. Would you like a concrete example using your current dispatch creation flow and the exact LiveKit Python client calls you use to join rooms? If so, share how you connect to rooms in the agent (sync vs async and what client functions you call). , with interactive elements

does await has any performance issue during dispatch?

await asyncio.wait_for(
                self.lkapi.agent_dispatch.create_dispatch(
                    CreateAgentDispatchRequest(
                        agent_name="outbound-call-agent",
                        room=room_name,
                        metadata=json.dumps(metadata)
                    )
                ),
                timeout=10.0
            )

Thanks — about performance for that

await asyncio.wait_for(...)

call: •

await

itself does not cause performance issues; it yields control while the HTTP request runs so your event loop can do other work. • Using

asyncio.wait_for(..., timeout=10.0)

is good practice to avoid hangs. • Sequentially awaiting many

create_dispatch

calls will be slower than issuing them concurrently. For better throughput, dispatch concurrently with

asyncio.gather()

or create tasks and await them, e.g.:

tasks = [
    self.lkapi.agent_dispatch.create_dispatch(
        CreateAgentDispatchRequest(agent_name="outbound-call-agent", room=r, metadata=json.dumps(metadata))
    )
    for r in rooms
]
results = await asyncio.gather(*tasks, return_exceptions=True)

• If your single container must handle multiple active dispatches, ensure the agent process is concurrency-capable (async handlers, semaphores to limit concurrency, or thread/executor for blocking work). • Keep per-dispatch timeouts/retries and monitor resource usage; for very high volume, batch or rate-limit dispatches or scale horizontally. If you want, I can: • Show a concurrent example with per-dispatch timeouts/retries, or • Convert your current

wait_for

pattern to a concurrent-safe version. Which would you like? Sources: Agent dispatch | LiveKit Documentation | Voice Agent Architecture: STT, LLM, and TTS Pipelines Explained | LiveKit, with interactive elements