how many agent we can have in one project

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

is it possible to write a toml file that deploy two agent in one project

You're asking whether a single

livekit.toml

can declare and deploy two agents in one project — good question. I'll address that and offer the practical alternatives. Short answer: • You cannot define two separate agent deployments in a single

livekit.toml

for CLI deployment.

livekit.toml

is scoped to a single agent deployment when used with

lk agent create

/

lk agent deploy

. What you can do instead: • Deploy multiple agents to the same project by creating one configuration (one

livekit.toml

) per agent and running separate

lk agent create

/

lk agent deploy

commands for each. Each agent config includes the project subdomain and its agent ID so they can share the same project credentials. • Give agents distinct names so you can explicitly dispatch them to rooms (named agents are not auto-dispatched). • Use agent dispatch rules or token-based

RoomAgentDispatch

entries to start or assign multiple agents to a room when needed. • Consider a multi-tenant agent design (one agent serving multiple users/tenants) if you want to scale without creating many deployments. Would you like: • An example

livekit.toml

for one agent so you can duplicate and adapt it for a second agent? • Guidance on naming/dispatch rules or an example token configuration to start multiple agents for a room? Sources: Server-side | livekit/livekit_composite | DeepWiki | Voice AI quickstart | LiveKit Documentation | Why do jobs stop after agent rejection? | LiveKit | Agent dispatch | LiveKit Documentation

• Guidance on naming/dispatch rules or an example token configuration to start multiple agents for a room?

Thanks — I’ll walk you through naming and dispatching multiple agents to the same room, and give concrete examples you can use. Quick overview • Give each agent an explicit name (set

agent_name

/

agentName

). Named agents are not auto-dispatched and must be explicitly assigned to rooms. • You can dispatch multiple named agents at once in three common ways: a. Embed

RoomConfiguration.agents

in the access token used to join/create the room. b. Create a SIP dispatch rule (for telephony) that includes multiple agents. c. Call the Agent Dispatch API or CLI at runtime to add multiple agents to an existing room. If you tell me which language/SDK (Node, Python, or CLI/SIP) you prefer, I can tailor the examples. Below are ready-to-use examples for token-based, API/CLI, and SIP dispatch. 1) Naming agents in your worker code • Python example:

@server.rtc_session(agent_name="transcriber-agent")
async def transcriber_agent(ctx: JobContext):
    ...

@server.rtc_session(agent_name="summarizer-agent")
async def summarizer_agent(ctx: JobContext):
    ...

• Node example:

const transcriberOpts = new WorkerOptions({ agentName: "transcriber-agent", ... });
const summarizerOpts = new WorkerOptions({ agentName: "summarizer-agent", ... });

Naming agents disables automatic dispatch and lets you control exactly when/where each agent runs. 2) Dispatch multiple agents via the access token (creates room and starts both agents) • Python (server SDK pattern):

from livekit.api import AccessToken, RoomAgentDispatch, RoomConfiguration, VideoGrants

token = (
  AccessToken()
    .with_identity("user-123")
    .with_grants(VideoGrants(room_join=True, room="my-room"))
    .with_room_config(
      RoomConfiguration(
        agents = [
          RoomAgentDispatch(agent_name="transcriber-agent", metadata='{"mode":"realtime"}'),
          RoomAgentDispatch(agent_name="summarizer-agent", metadata='{"priority":"high"}'),
        ]
      )
    )
    .to_jwt()
)

• Node (server SDK pattern):

import { RoomAgentDispatch, RoomConfiguration } from '@livekit/protocol';
import { AccessToken } from 'livekit-server-sdk';

const at = new AccessToken();
at.identity = 'user-123';
at.addGrant({ roomJoin: true, room: 'my-room' });
at.roomConfig = new RoomConfiguration({
  agents: [
    new RoomAgentDispatch({ agentName: 'transcriber-agent', metadata: '{"mode":"realtime"}' }),
    new RoomAgentDispatch({ agentName: 'summarizer-agent', metadata: '{"priority":"high"}' }),
  ],
});
const token = await at.toJwt();

This token, when used to join or create the room, will dispatch both agents to the room and each agent receives its

JobContext

with the provided metadata. 3) Dispatch dynamically via API or CLI (add agents to an existing room) • CLI:

lk dispatch create --agent-name transcriber-agent --room my-room --metadata '{"mode":"realtime"}'
lk dispatch create --agent-name summarizer-agent --room my-room --metadata '{"priority":"high"}'

• Python (API call pattern):

await lkapi.agent_dispatch.create_dispatch(
  api.CreateAgentDispatchRequest(room="my-room", agent_name="transcriber-agent", metadata='{"mode":"realtime"}')
)
await lkapi.agent_dispatch.create_dispatch(
  api.CreateAgentDispatchRequest(room="my-room", agent_name="summarizer-agent", metadata='{"priority":"high"}')
)

Use this when you want to trigger agents conditionally or after the room is already running. 4) SIP dispatch rule (for inbound calls) Create a dispatch JSON that includes multiple agents:

{
  "dispatch_rule": {
    "rule": {
      "dispatchRuleIndividual": {
        "roomPrefix": "call-"
      }
    },
    "roomConfig": {
      "agents": [
        {
          "agentName": "voice-assistant",
          "metadata": "{\"intent\":\"support\"}"
        },
        {
          "agentName": "transcriber",
          "metadata": "{\"format\":\"text\"}"
        }
      ]
    }
  }
}

Deploy with:

lk sip dispatch create dispatch-rule.json

Incoming SIP calls that match the rule will create a room and dispatch both agents. Best practices and tips • Use clear agent names (e.g.,

transcriber-agent

,

summarizer-agent

) so your orchestration and logs are easy to follow. • Pass useful JSON in

metadata

per agent so each agent receives the context it needs. • If you expect a high number of users, prefer multi-tenant agent patterns (one agent instance serving many users) or ensure your plan’s concurrent agent session limits meet your needs. • If several worker instances share the same

agent_name

, jobs will be load-balanced across them. Would you like: • A copyable token example for your exact SDK keys/room name, or • A sample

lk

CLI sequence and

livekit.toml

snippet for deploying two named agents? Tell me which option (Node, Python, CLI/SIP) you want a ready-to-run snippet for and I’ll generate it. Sources: Agents telephony integration | LiveKit Documentation | Agent dispatch | LiveKit Documentation | Agents Framework | livekit/livekit_composite | DeepWiki | Why do jobs stop after agent rejection? | LiveKit | server-sdk-go | livekit/livekit_composite | DeepWiki

do you have an example of calling remote-graph and streaming messages

Thanks — I’m responding to your request for an example of calling “remote-graph” and streaming messages. I don’t have any reference for a specific "remote-graph" API in the LiveKit sources. However, I can help if you meant one of these related capabilities: • RPC between participants (calling remote functions on other participants in a room). • Streaming data (audio or binary) to a remote processor using ByteStreams or data channels. • Agent message streaming patterns (agents producing/consuming streaming text or audio via async iteration and data channels). Which of the above matches what you mean by “remote-graph”? Or if “remote-graph” is a particular external service or library (for example a graph-based orchestration or LangGraph-style service), tell me: • Which SDK/language you want examples for (Node.js or Python). • Whether you need audio streaming, text streaming, or bidirectional message streaming. • Whether you want the stream triggered from a participant, an agent, or the server. If you confirm one of those options, I’ll provide a concrete, copy-pastable example showing: • How to invoke the remote endpoint (RPC or ByteStream). • How to stream messages progressively. • How to handle cancellation and progressive responses in the client and agent code. Sources: Bringing AI avatars to voice agents | examples/other | livekit/livekit_composite | DeepWiki | livekit/livekit_composite | DeepWiki

running something like session = AgentSession( stt=inference.STT( model="elevenlabs/scribe_v2_realtime", language="en", ), llm=build_remote_langgraph_llm(job_id=job_id, room_name=room_name), tts=inference.TTS( model="elevenlabs/eleven_flash_v2_5", # use flash model for faster response times, or turbo for higher quality voice="hpp4J3VqNfWAUOO0d1Us", #hpp4J3VqNfWAUOO0d1Us #Xb7hH8MSUJpSbSDYk0k2 language="en" ), turn_detection=EnglishModel(), vad=vad, # With remote LangGraph thread memory, speculative turns can get persisted and replayed. preemptive_generation=False, )

Thanks — I see you’re building an AgentSession that wires STT, a remote LangGraph-backed LLM, TTS, VAD, and turn detection. I don’t have a concrete reference for a built-in “remote-graph” or LangGraph client in the LiveKit sources, but I can show a safe, practical Python pattern you can adapt: treat the remote LangGraph LLM as a separate async streaming service (RPC/HTTP/WebSocket), and connect it into the AgentSession via an async generator that sends partial STT chunks and yields incremental responses back to the agent for TTS and room delivery. Below is a high-level, copy-pastable Python example pattern you can adapt. It focuses on: • Receiving streaming STT output from the STT component. • Forwarding partial transcripts to a remote LLM over a streaming WebSocket (or async HTTP) connection. • Receiving incremental responses from the remote LLM and passing them into the agent’s output path for TTS and room messaging. • Handling cancellation and graceful shutdown. Note: replace the placeholder remote LLM client code with your actual LangGraph client or RPC calls.

import asyncio
import json
from typing import AsyncGenerator

# --- Placeholder remote LLM streaming client ---
# Replace with your LangGraph/WebSocket/HTTP streaming client.
class RemoteLLMClient:
    def __init__(self, endpoint: str, job_id: str, room_name: str):
        self.endpoint = endpoint
        self.job_id = job_id
        self.room_name = room_name
        self._ws = None

    async def connect(self):
        # Example: open a websocket connection to remote LLM
        import websockets
        self._ws = await websockets.connect(self.endpoint)
        # send an initial handshake
        await self._ws.send(json.dumps({"type": "handshake", "job_id": self.job_id, "room": self.room_name}))

    async def send_partial(self, text_chunk: str):
        # send a partial transcript chunk
        await self._ws.send(json.dumps({"type": "input_chunk", "text": text_chunk}))

    async def responses(self) -> AsyncGenerator[str, None]:
        # yield incremental responses from the remote LLM
        try:
            async for raw in self._ws:
                msg = json.loads(raw)
                if msg.get("type") == "token":
                    yield msg.get("token_text", "")
                elif msg.get("type") == "done":
                    break
        finally:
            await self.close()

    async def close(self):
        if self._ws:
            await self._ws.close()
            self._ws = None

# --- Agent session integration pattern ---
async def run_agent_session(job_ctx, room_name: str, job_id: str):
    """
    job_ctx: the agent's JobContext provided by LiveKit agent runtime (placeholder)
    room_name / job_id: identifiers used for remote LLM
    """
    # Create the remote LLM client and connect
    llm = RemoteLLMClient(endpoint="<wss://llm.example/stream>", job_id=job_id, room_name=room_name)
    await llm.connect()

    # Queue for STT partials coming from STT pipeline
    stt_queue: asyncio.Queue[str] = asyncio.Queue()
    # Queue for assembled LLM output tokens/chunks to forward to TTS / room
    llm_out_queue: asyncio.Queue[str] = asyncio.Queue()

    # Task: consume STT partials and forward to remote LLM
    async def stt_forwarder():
        while True:
            chunk = await stt_queue.get()
            if chunk is None:
                break
            await llm.send_partial(chunk)
        # signal end of input
        await llm.send_partial("")  # or send a dedicated "eof" message if your LLM expects it)

    # Task: consume remote LLM responses and place into llm_out_queue
    async def llm_reader():
        async for token in llm.responses():
            await llm_out_queue.put(token)
        await llm_out_queue.put(None)  # sentinel for EOF

    # Task: consume llm_out_queue and forward to TTS and/or data-channel publish
    async def output_worker():
        buffer = []
        while True:
            token = await llm_out_queue.get()
            if token is None:
                break
            buffer.append(token)
            partial_text = "".join(buffer)

            # 1) Send incremental text message to the room (via job_ctx API)
            # job_ctx.publish_message or similar — replace with actual call
            try:
                await job_ctx.publish_message({"type": "assistant.partial", "text": partial_text})
            except Exception:
                pass

            # 2) Optionally drive TTS for progressive audio playback
            # If using TTS that supports streaming, feed `token` or `partial_text` into it progressively.
            # Example: job_ctx.tts_stream.send(partial_text)
            # Or buffer up and call TTS at sentence boundaries.

        # On EOF, send final assembled response
        final_text = "".join(buffer)
        await job_ctx.publish_message({"type": "assistant.final", "text": final_text})

    # Start tasks
    tasks = [
        asyncio.create_task(stt_forwarder()),
        asyncio.create_task(llm_reader()),
        asyncio.create_task(output_worker())
    ]

    # Example: hook STT stream from job_ctx to stt_queue
    # This depends on your STT API: it might provide an async generator of partial transcripts.
    try:
        async for stt_chunk in job_ctx.stt_stream():  # placeholder API
            if stt_chunk is None:
                break
            await stt_queue.put(stt_chunk)
    except asyncio.CancelledError:
        pass
    finally:
        # signal finish to forwarder
        await stt_queue.put(None)

    # Wait for workers to finish
    await asyncio.gather(*tasks, return_exceptions=True)
    await llm.close()

How this maps to your snippet: • Use your

inference.STT

to supply incremental transcripts into

stt_queue

(replace

job_ctx.stt_stream()

with the real STT streaming API you have). • Replace

build_remote_langgraph_llm(...)

with a client that implements the pattern shown (connect, send partials, and async iterate responses). • Use

inference.TTS

to either stream audio progressively or synthesize final audio from

final_text

. If your TTS supports streaming, call it from

output_worker

as tokens arrive. Practical tips • Choose whether to stream TTS progressively (lower latency) or synthesize at sentence/turn end (simpler). • Use metadata in dispatch/token to let the remote LLM know context (job_id, room_name, user info). • If your remote LLM supports delta tokens, forward them immediately to improve responsiveness. • Implement backpressure: if the remote LLM is slow, consider buffering limits and drop/merge low-value partials. • Ensure you handle cancellation and connection errors (cleanly close websockets). If you want, I can: • Turn this into a concrete example using your actual STT/TTS SDK calls and the exact LangGraph client you have, if you share the LangGraph endpoint or client library you plan to use. • Provide a Node.js equivalent using WebSocket streams. Which would you prefer?

what should I put here

# --- Placeholder remote LLM streaming client ---
# Replace with your LangGraph/WebSocket/HTTP streaming client.

Great — I’ll show a concrete, practical implementation you can drop into that placeholder. Below are two ready-to-use patterns for a remote LLM streaming client you can adapt to your LangGraph endpoint: (A) WebSocket-based streaming (common for incremental token streams), and (B) HTTP chunked/Server-Sent-Events streaming (if your service uses HTTP). Choose whichever matches your LangGraph service. Both examples implement: • connect() • send_partial(text_chunk) • responses() — an async generator that yields incremental text tokens/chunks • close() A) WebSocket-based remote LLM client (recommended for low-latency incremental tokens)

import asyncio
import json
from typing import AsyncGenerator, Optional

import websockets  # pip install websockets

class RemoteLLMClient:
    def __init__(self, endpoint: str, job_id: str, room_name: str, api_key: Optional[str] = None):
        self.endpoint = endpoint
        self.job_id = job_id
        self.room_name = room_name
        self.api_key = api_key
        self._ws = None
        self._recv_task = None
        self._queue: asyncio.Queue[Optional[str]] = asyncio.Queue()

    async def connect(self):
        headers = {}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
        self._ws = await websockets.connect(self.endpoint, extra_headers=headers)
        # optional handshake, include job/room context
        await self._ws.send(json.dumps({"type": "handshake", "job_id": self.job_id, "room": self.room_name}))
        # start reader loop
        self._recv_task = asyncio.create_task(self._reader())

    async def _reader(self):
        try:
            async for raw in self._ws:
                try:
                    msg = json.loads(raw)
                except Exception:
                    # treat raw text as token if not JSON
                    await self._queue.put(raw)
                    continue

                # adjust to your remote schema: token/delta/done
                typ = msg.get("type")
                if typ == "token" or typ == "delta":
                    token_text = msg.get("text") or msg.get("token_text") or ""
                    await self._queue.put(token_text)
                elif typ == "done":
                    break
                elif typ == "error":
                    await self._queue.put(None)
                    break
            # signal EOF
            await self._queue.put(None)
        except Exception:
            await self._queue.put(None)

    async def send_partial(self, text_chunk: str):
        # Decide how to package partials for your service
        if self._ws is None:
            raise RuntimeError("Not connected")
        payload = {"type": "input_chunk", "text": text_chunk, "job_id": self.job_id}
        await self._ws.send(json.dumps(payload))

    async def responses(self) -> AsyncGenerator[str, None]:
        while True:
            token = await self._queue.get()
            if token is None:
                break
            yield token

    async def close(self):
        if self._ws:
            try:
                await self._ws.close()
            except Exception:
                pass
            self._ws = None
        if self._recv_task:
            self._recv_task.cancel()
            self._recv_task = None

How to adapt: • Replace the handshake and message fields with the exact JSON your LangGraph server expects. • If LangGraph requires a session token or headers, pass them via

api_key

or

extra_headers

. • If your service streams tokens under a different field name, update

_reader()

parsing accordingly. B) HTTP streaming (chunked responses or SSE)

import aiohttp
import asyncio
import json
from typing import AsyncGenerator, Optional

class RemoteLLMHTTPClient:
    def __init__(self, endpoint: str, job_id: str, room_name: str, api_key: Optional[str] = None):
        self.endpoint = endpoint
        self.job_id = job_id
        self.room_name = room_name
        self.api_key = api_key
        self._session = aiohttp.ClientSession()

    async def connect(self):
        # No persistent connection required; connect can be a noop or a handshake POST
        pass

    async def send_partial(self, text_chunk: str):
        # Many HTTP-based services accept incremental inputs via POST to an "append" endpoint.
        # If your LangGraph supports this, use it; otherwise, you can buffer locally and send on finalization.
        url = f"{self.endpoint}/append"
        headers = {}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
        payload = {"job_id": self.job_id, "room": self.room_name, "text": text_chunk}
        async with <http://self._session.post|self._session.post>(url, json=payload, headers=headers) as resp:
            resp.raise_for_status()
            return await resp.text()

    async def responses(self) -> AsyncGenerator[str, None]:
        # This example assumes the server offers a streaming endpoint that returns a chunked JSON stream.
        url = f"{self.endpoint}/stream?job_id={self.job_id}&room={self.room_name}"
        headers = {}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
        async with self._session.get(url, headers=headers) as resp:
            resp.raise_for_status()
            async for chunk, _ in resp.content.iter_chunks():
                if not chunk:
                    continue
                text = chunk.decode()
                # If chunk contains JSON messages, parse them, else treat as raw token text
                try:
                    msg = json.loads(text)
                    if msg.get("type") in ("token", "delta"):
                        yield msg.get("text") or msg.get("token_text", "")
                    elif msg.get("type") == "done":
                        break
                except Exception:
                    # non-json chunk
                    yield text
        return

    async def close(self):
        await self._session.close()

Integration notes and best practices • Keep tokens small and frequent for low-latency experience; send punctuation/pauses as boundaries for sending to TTS. • Use metadata in handshake (job_id, room, user_id) so LangGraph can load or persist thread memory. • Implement backpressure: drop or coalesce STT partials if the remote LLM cannot keep up. • Ensure graceful close and error handling so your agent can retry or fall back. • If your LangGraph client library already exists, wrap its stream API to implement the same connect/send_partial/responses interface shown above. If you share: • the exact LangGraph endpoint/protocol (WebSocket, SSE, or an SDK), or • a sample message schema (what the LLM sends for incremental tokens), I’ll adapt this example to exactly match and produce a ready-to-run snippet wired into your AgentSession. Which do you prefer (WebSocket or HTTP/SSE), and can you paste one example message your LangGraph server returns?