Acknowledged — you want a minimal, self-contained Python example showing: • an agent server that can register a dynamic agent name and connect to LiveKit, and • an API endpoint that, on request, reads agent data (simulated DB), creates/uses a room, and dispatches the agent into that room. Below are two simple files you can run locally. They use: • FastAPI for the HTTP endpoint, • an async LiveKit client-like sketch that calls the AgentDispatchService

create_dispatch

API (illustrative; adapt to your SDK/client), • a simulated database lookup (dictionary). Save both files, install dependencies (

pip install fastapi uvicorn httpx

), then run

uvicorn api:app --reload

and run the agent server in another process. 1) agent_server.py — simple agent process that pretends to register/start an agent and listens for join requests (illustrative)

# agent_server.py
import asyncio
import uuid
import httpx

LIVEKIT_BASE = "<http://localhost:7880>"   # replace with your LiveKit control plane endpoint
AGENT_NAME = "dynamic-agent-1"           # agent name will be set when started

async def register_agent(agent_name: str):
    # Illustrative: register agent with your control plane if needed.
    print(f"[agent] registering agent: {agent_name}")
    # If your deployment requires informing LiveKit control plane or a custom registry, do it here.
    await asyncio.sleep(0.1)
    print(f"[agent] registered")

async def run_agent_loop(agent_name: str):
    print(f"[agent] starting main loop for {agent_name}")
    # In a real agent, connect to LiveKit as a participant, handle audio, websocket events, etc.
    # Here we simply poll a (fictional) local status endpoint to demonstrate lifecycle.
    while True:
        # check for work / room assignment (placeholder)
        await asyncio.sleep(5)
        print(f"[agent] ({agent_name}) heartbeat")

async def main():
    # allow unique agent names easily for dynamic creation
    unique_name = f"{AGENT_NAME}-{uuid.uuid4().hex[:6]}"
    await register_agent(unique_name)
    await run_agent_loop(unique_name)

if __name__ == "__main__":
    asyncio.run(main())

2) api.py — FastAPI endpoint that simulates fetching agent data from DB, creates a room name, and calls the AgentDispatchService create_dispatch endpoint.

# api.py
from fastapi import FastAPI, HTTPException
import asyncio
import httpx
import uuid

app = FastAPI()

# Simulated DB of agents
AGENT_DB = {
    "1": {"agent_name": "dynamic-agent-1", "metadata": {"skill": "sales"}},
    "2": {"agent_name": "dynamic-agent-2", "metadata": {"skill": "support"}},
}

# Replace with your LiveKit control plane / agent-dispatch API base and auth
LIVEKIT_API_BASE = "<http://localhost:7880>"   # control-plane base URL for agent dispatch API
API_KEY = "example-key"                      # use proper auth in prod

async def create_dispatch(agent_name: str, room: str, metadata: dict):
    """
    Illustrative call to AgentDispatchService.create_dispatch.
    Adjust request path, body, and auth to match your LiveKit control plane API/SDK.
    """
    url = f"{LIVEKIT_API_BASE}/v1/agent_dispatch"  # illustrative path
    payload = {
        "agent_name": agent_name,
        "room": room,
        "metadata": metadata,
    }
    headers = {"Authorization": f"Bearer {API_KEY}"}
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await <http://client.post|client.post>(url, json=payload, headers=headers)
        if resp.status_code >= 400:
            raise HTTPException(status_code=500, detail=f"dispatch failed: {resp.text}")
        return resp.json()

@app.post("/dispatch/{agent_id}")
async def dispatch_agent(agent_id: str):
    """
    API endpoint: fetch agent info, ensure room, and create a dispatch.
    """
    agent = AGENT_DB.get(agent_id)
    if not agent:
        raise HTTPException(status_code=404, detail="agent not found")

    # Create a unique room name (or use business logic)
    room_name = f"room-{uuid.uuid4().hex[:8]}"

    # Optionally: create room ahead of dispatch using LiveKit Rooms API
    # For many use-cases create_dispatch will create the room if it doesn't exist.

    # Call agent dispatch
    dispatch_resp = await create_dispatch(agent_name=agent["agent_name"], room=room_name, metadata=agent["metadata"])

    return {"room": room_name, "dispatch": dispatch_resp}

Notes and things to adapt to your environment: • The code above is illustrative: replace

LIVEKIT_API_BASE

,

API_KEY

, and API paths with your LiveKit control plane's actual endpoints or SDK calls. • The LiveKit AgentDispatchService can create the room and start the named agent via a

create_dispatch

call; the example shows how to call that endpoint. If you use the official Python SDK, call the SDK's

create_dispatch

method instead of HTTP. • If your agent server must match the named agent exactly, ensure the agent process uses the same

agent_name

as dispatched. For dynamic agents you can generate unique names and store them in DB so the API uses that same name when dispatching. • If you want the room created automatically when dispatching,

create_dispatch

will create it if absent; otherwise you can call the Rooms API first. If you’d like, I can: • Convert the illustrative HTTP call to the actual LiveKit Python SDK method call, • Show how to include room configuration or SIP grants in the token/dispatch, • Add an example where the agent server registers its name dynamically in a simple registry the API reads. Which of those would you like next? Sources: SIP dispatch rule | LiveKit docs | Agent dispatch | LiveKit docs | Authentication | LiveKit docs | Agents telephony integration | LiveKit docs