hi

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

how can i jetc dynamic data ot ajent like nmae of th euse and nametime timz zonee etc

Thanks — I can help with that. Just to confirm, are you asking how to send dynamic user info (name, timezone, etc.) to a LiveKit agent when you start/dispatch it? If yes, which of these are you using? • Dispatching an agent via the server API/CLI • Starting an AgentSession from your app (Agent SDK) • Participant tokens / joining a room and passing metadata Answer that and I’ll give a short code example for the exact method. Sources: Agent dispatch | LiveKit docs | Context Variables | LiveKit docs

i need to inject runtim data to to agent

Thanks — I’ll help you inject runtime data into the agent. Quick clarification so I can give the exact code: which method are you using to start the agent? • Dispatching an agent via the server API/CLI (use job metadata) • Starting an AgentSession from your app (use context variables on the session) • Passing data via participant tokens when joining a room Reply with the method and the language/SDK you’re using (Node, Python, or other) and I’ll provide a concise example.

import json import httpx from dotenv import load_dotenv from livekit.agents import Agent, JobContext, JobProcess, WorkerOptions, cli, function_tool from livekit.agents.voice import AgentSession from livekit.plugins import silero from agents.utils import get_llm_from_config, get_stt_from_config, get_tts_from_config from agents.utils import WELCOME_MESSAGE, DEFAULT_SYSTEM_PROMPT from agents.utils.config import get_all_metadata from agents.functions.caldotcom import get_free_slots, book_slot load_dotenv() def prewarm(proc: JobProcess): proc.userdata.update({ "vad": silero.VAD.load(), }) class V1TestAgent(Agent): def __init__(self, meta_data): agent_info = meta_data['agent'] super().__init__( instructions=agent_info.get("system_prompt", DEFAULT_SYSTEM_PROMPT), stt=get_stt_from_config(meta_data['stt']), llm=get_llm_from_config(meta_data['llm']), tts=get_tts_from_config(meta_data['tts']) ) @function_tool() async def list_event_types(self, start_time: str, end_time: str) -> str: """ Fetch available slots for a given date/time range. Parameters should be ISO 8601 strings, e.g., "2025-12-29T000000Z" """ slots = await get_free_slots(start_time, end_time) return json.dumps(slots, indent=2) @function_tool() async def book_available_slot(self, start_time_utc: str, name: str, email: str) -> str: """ timezone="America/New_York" Book a slot using Cal.com API. """ timezone="America/New_York" eventTypeId=4264613 booking = await book_slot(start_time_utc, name, email, timezone, eventTypeId) if booking is None: return json.dumps({"error": "Failed to book slot"}, indent=2) return json.dumps(booking, indent=2) async def entrypoint(ctx: JobContext): meta_data = await get_all_metadata(ctx) agent = V1TestAgent(meta_data) session = AgentSession() await session.start(agent=agent, room=ctx.room) if True: await session.say(meta_data['agent'].get('greeting_message', WELCOME_MESSAGE)) async def on_session_end(): ctx.delete_room(ctx.room.name) ctx.add_shutdown_callback(on_session_end) def run(): cli.run_app( WorkerOptions( entrypoint_fnc=entrypoint, prewarm_fnc=prewarm, agent_name="v1_test_agent", port=8082, ) ) if name == "__main__": run() in this

Thanks — I see your Python agent code and that you want to inject runtime data (user name, timezone, etc.) into the agent when the session starts. You can pass that data either as job metadata when dispatching the job or as context variables on the AgentSession. Below are concise examples for both and the small changes to use them in your code. 1) Using job metadata (dispatch -> available in

ctx.job.metadata

) • When dispatching the agent, include JSON in

metadata

, e.g.:

{
  "user_name": "Alice",
  "timezone": "America/New_York",
  "email": "<mailto:alice@example.com|alice@example.com>"
}

• In your entrypoint you already call `get_all_metadata(ctx)`; if that returns parsed metadata, use it like:

meta_data = await get_all_metadata(ctx)
user_name = meta_data.get("user_name")
timezone = meta_data.get("timezone")

• Pass into the agent or use directly when starting session:

agent = V1TestAgent(meta_data)
session = AgentSession()
await session.start(agent=agent, room=ctx.room)
await session.say(meta_data.get('greeting_message', WELCOME_MESSAGE).replace("{user}", user_name))

2) Using AgentSession context variables (set at session start) • Provide a dict of

context

when starting the session:

context_vars = {"user_name": "Alice", "timezone": "America/New_York", "email":"<mailto:alice@example.com|alice@example.com>"}
await session.start(agent=agent, room=ctx.room, context=context_vars)

• Inside your agent tools/LLM prompts, reference

session.context

(or access passed context from the AgentSession API). If your agent constructor needs these values, pass them in:

agent = V1TestAgent({**meta_data, **context_vars})

(or extract and pass specific fields) 3) Minor code changes for your snippet • Ensure you parse metadata JSON if needed and merge any runtime context:

async def entrypoint(ctx: JobContext):
    meta_data = await get_all_metadata(ctx)  # make sure this returns a dict
    runtime = {"user_name": "Alice", "timezone": "America/New_York"}  # replace with actual runtime values
    merged = {**meta_data, **runtime}
    agent = V1TestAgent(merged)
    session = AgentSession()
    await session.start(agent=agent, room=ctx.room, context=runtime)
    await session.say(merged['agent'].get('greeting_message', WELCOME_MESSAGE))

Notes and tips • Put user-specific values in top-level metadata fields (not only inside

agent

) so your agent code can easily access them. • Use JSON-formatted metadata when dispatching; parse it to a dict in

entrypoint

. • Choose job metadata when information is set at dispatch time; choose session

context

when data is known only at session start or needs to be ephemeral. If you tell me how you dispatch the job (server API, CLI, or participant token) and which fields you want passed at runtime, I’ll give the exact dispatch payload and the minimal code edit to inject them.

on run time

Thanks — you want to inject runtime data into the agent while it runs. You can do that with AgentSession context (best for data known at session start) — here's the minimal change to your entrypoint and how to pass runtime values at start time. 1) Entry point — accept and merge runtime context, pass to session and agent:

async def entrypoint(ctx: JobContext):
    meta_data = await get_all_metadata(ctx)  # dict from dispatch-time metadata

    # runtime values (replace with real runtime sources)
    runtime = {
        "user_name": "Alice",
        "timezone": "America/New_York",
        "email": "<mailto:alice@example.com|alice@example.com>"
    }

    # merge so agent sees both dispatch metadata and runtime data
    merged = {**meta_data, **runtime}

    agent = V1TestAgent(merged)
    session = AgentSession()

    # pass runtime into the running session via context
    await session.start(agent=agent, room=ctx.room, context=runtime)

    await session.say(merged['agent'].get('greeting_message', WELCOME_MESSAGE))

2) Use inside agent and tools • In your agent code, read from the metadata you passed into the constructor (e.g.,

meta_data['user_name']

) or from session-provided context if your AgentSession exposes it to tools. Example in a tool:

@function_tool()
async def book_available_slot(self, start_time_utc: str, name: str, email: str) -> str:
    # use merged metadata stored on the agent instance
    timezone = self.meta.get("timezone", "UTC")
    # ...

(If your V1TestAgent stores

meta_data

on

self.meta

, access it there.) 3) Where runtime values come from • If data is available at the moment the session starts (e.g., web request, participant join, or UI form), populate the

runtime

dict from that source and pass it in

session.start(..., context=runtime)

. • If dispatch-time only: include those fields in the

ctx.job.metadata

JSON and read them via

get_all_metadata(ctx)

. If you tell me which runtime source you will use (HTTP request, participant token, UI form, or webhook), I’ll give the exact one-line change to populate

runtime

from that source.

i wan tadd int to ingant intruincs

Got it — you want to add runtime intent data directly into the agent’s instructions. I’ll show the minimal, practical change for your Python agent so an intent (name + params) is injected into the system prompt at session start and available to tools. Quick question: do you want the intent set once at session start, or changed mid-session? (reply "start" or "mid") If "start", add this to your entrypoint:

async def entrypoint(ctx: JobContext):
    meta_data = await get_all_metadata(ctx)  # dict from dispatch-time metadata

    # runtime intent values (replace with real runtime source)
    runtime = {
        "intent_name": "book_meeting",
        "intent_params": {"duration_minutes": 30, "preferred_time": "2025-12-30T15:00:00Z"},
        "user_name": "Alice",
        "timezone": "America/New_York"
    }

    # build a dynamic system prompt including the intent
    system_instructions = meta_data['agent'].get(
        "system_prompt",
        "You are a helpful assistant."
    )
    dynamic_instructions = (
        f"{system_instructions}\n\n"
        f"Runtime intent: {runtime['intent_name']}\n"
        f"Parameters: {json.dumps(runtime['intent_params'])}\n"
        f"User: {runtime['user_name']} (timezone: {runtime['timezone']})"
    )

    # pass merged metadata into the agent
    merged = {**meta_data, **runtime}
    agent = V1TestAgent(merged)
    session = AgentSession()

    # pass runtime context to the running session
    await session.start(agent=agent, room=ctx.room, context=runtime)

    # send the greeting using merged data
    await session.say(merged['agent'].get('greeting_message', WELCOME_MESSAGE))

And inside your agent, store/access the merged metadata and use the dynamic instructions: • In V1TestAgent.init, accept and save meta:

class V1TestAgent(Agent):
    def __init__(self, meta_data):
        self.meta = meta_data
        instructions = meta_data.get('agent', {}).get('system_prompt') or DEFAULT_SYSTEM_PROMPT
        # If we included dynamic_instructions under meta (as above), use that:
        if meta_data.get('intent_name'):
            instructions = (
                f"{instructions}\n\n"
                f"Runtime intent: {meta_data['intent_name']}\n"
                f"Params: {json.dumps(meta_data.get('intent_params', {}))}"
            )
        super().__init__(instructions=instructions, ...)

• In tools, read `self.meta`:

@function_tool()
async def book_available_slot(self, start_time_utc: str, name: str, email: str) -> str:
    timezone = self.meta.get("timezone", "UTC")
    intent = self.meta.get("intent_name")
    params = self.meta.get("intent_params", {})
    # use intent/params to influence booking logic

If you want to change intent mid-session, tell me and I’ll provide the small code to call

update_instructions()

or to push a new chat message into the session context so the LLM re-weights behavior immediately. Sources: Context Variables | LiveKit docs | Job lifecycle | LiveKit docs | Agent dispatch | LiveKit docs | External data and RAG | LiveKit docs | Building an Automated IVR Menu Caller | LiveKit docs

import json import httpx from dotenv import load_dotenv from livekit.agents import Agent, JobContext, JobProcess, WorkerOptions, cli, function_tool from livekit.agents.voice import AgentSession from livekit.plugins import silero from agents.utils import get_llm_from_config, get_stt_from_config, get_tts_from_config from agents.utils import WELCOME_MESSAGE, DEFAULT_SYSTEM_PROMPT from agents.utils.config import get_all_metadata from agents.functions.caldotcom import get_free_slots, book_slot load_dotenv() def prewarm(proc: JobProcess): proc.userdata.update({ "vad": silero.VAD.load(), }) class V1TestAgent(Agent): def __init__(self, meta_data): agent_info = meta_data['agent'] super().__init__( instructions=agent_info.get("system_prompt", DEFAULT_SYSTEM_PROMPT), stt=get_stt_from_config(meta_data['stt']), llm=get_llm_from_config(meta_data['llm']), tts=get_tts_from_config(meta_data['tts']) ) async def on_enter(self): new_instructions = f""" You are a helpful scheduling assistant. The current user is arun (email: arun@arun.com) in timezone America/New_York. Always ask the user to confirm date/time before booking. """ await self.update_instructions(new_instructions) @function_tool() async def list_event_types(self, start_time: str, end_time: str) -> str: """ Fetch available slots for a given date/time range. Parameters should be ISO 8601 strings, e.g., "2025-12-29T000000Z" """ slots = await get_free_slots(start_time, end_time) return json.dumps(slots, indent=2) @function_tool() async def book_available_slot(self, start_time_utc: str, name: str, email: str) -> str: """ timezone="America/New_York" Book a slot using Cal.com API. """ timezone="America/New_York" eventTypeId=4264613 booking = await book_slot(start_time_utc, name, email, timezone, eventTypeId) if booking is None: return json.dumps({"error": "Failed to book slot"}, indent=2) return json.dumps(booking, indent=2) async def entrypoint(ctx: JobContext): meta_data = await get_all_metadata(ctx) agent = V1TestAgent(meta_data) session = AgentSession() await session.start(agent=agent, room=ctx.room) if True: await session.say(meta_data['agent'].get('greeting_message', WELCOME_MESSAGE)) async def on_session_end(): ctx.delete_room(ctx.room.name) ctx.add_shutdown_callback(on_session_end) def run(): cli.run_app( WorkerOptions( entrypoint_fnc=entrypoint, prewarm_fnc=prewarm, agent_name="v1_test_agent", port=8082, ) ) if name == "__main__": run() when i say to agent byy by or cut the call it shoul end cll ait self

You're asking how to have the agent end the call itself when the user says "bye" or when the call is cut. Below are minimal, practical changes to your code to detect a goodbye and end the session cleanly. Choice to make: Do you want the agent to end the call when: • A user voice/text message contains a goodbye phrase (like "bye", "goodbye", "thanks, bye")? — recommended • The participant disconnects (call cut)? — handled by room/participant events Tell me if you want both; otherwise I’ll assume you want the agent to end on a goodbye phrase. 1) End on spoken/written goodbye (recommended) • Add a short check in your agent so when a user message arrives it looks for goodbye phrases and calls

await self.session.stop()

(or

await session.stop()

from entrypoint context). Example tool / message handler:

# inside your Agent subclass
GOODBYES = {"bye", "goodbye", "see you", "thanks, bye", "thank you, bye"}

async def on_message(self, message: str, *args, **kwargs):
    text = message.strip().lower()
    # simple substring check (can improve with regex / NLP)
    if any(g in text for g in GOODBYES):
        await self.say("Goodbye — ending the call now. Goodbye!")
        # stop the AgentSession driving this agent
        await self.session.stop()
        return
    # otherwise normal handling

• If your code uses the session/LLM pipeline rather than an

on_message

hook, create a small system-level tool that runs after each user message and performs the same check before proceeding. 2) Handle participant disconnect (call cut) • In the entrypoint, add a shutdown callback that watches participant presence and ends the session if the user leaves. Example pattern:

async def entrypoint(ctx: JobContext):
    meta_data = await get_all_metadata(ctx)
    agent = V1TestAgent(meta_data)
    session = AgentSession()

    await session.start(agent=agent, room=ctx.room)

    async def on_participant_update():
        # if the user participant left, stop the session
        if not ctx.room.get_participants():  # replace with your participant-check
            await session.stop()

    ctx.add_shutdown_callback(on_participant_update)

    await session.say(meta_data['agent'].get('greeting_message', WELCOME_MESSAGE))

• If your runtime exposes participant events, listen for the participant-left event and call

await session.stop()

. 3) Important details • Use

await session.stop()

to gracefully stop the session; it will close the room/connection and allow your

on_session_end

cleanup to run. • Send a final spoken/text goodbye via

await session.say(...)

before calling

stop()

for a polite exit. • For more robust detection use a small regex or intent classifier rather than simple substring matching. If you want, I can: • Add a concrete snippet wired to your existing

AgentSession

and

on_session_end

flow, or • Replace the simple substring detection with a regex or small intent-check function. Which would you prefer?

use fucniton tool calls

Thanks — I’ll show a minimal change to your code that uses a

@function_tool

to detect goodbye intents and end the call gracefully. I’ll also show how to inject runtime intent/data into instructions at session start so the tool can use it. Apply these edits: 1) Add a goodbye-check tool and a helper to stop the session

from livekit.agents import function_tool

GOODBYE_KEYWORDS = {"bye", "goodbye", "see you", "thanks", "thank you", "bye!"}

class V1TestAgent(Agent):
    def __init__(self, meta_data):
        self.meta = meta_data
        agent_info = meta_data['agent']
        super().__init__(
            instructions=agent_info.get("system_prompt", DEFAULT_SYSTEM_PROMPT),
            stt=get_stt_from_config(meta_data['stt']),
            llm=get_llm_from_config(meta_data['llm']),
            tts=get_tts_from_config(meta_data['tts'])
        )
        # session will be set when AgentSession starts this agent
        self.session = None

    async def on_enter(self):
        # keep existing behavior if you want
        new_instructions = f"""
You are a helpful scheduling assistant.
The current user is {self.meta.get('user_name', 'user')} (email: {self.meta.get('email', '')}) in timezone {self.meta.get('timezone', 'UTC')}.
Always ask the user to confirm date/time before booking.
"""
        await self.update_instructions(new_instructions)

    @function_tool()
    async def check_goodbye_and_maybe_end(self, user_text: str) -> str:
        """
        Returns "continue" if conversation should continue,
        Returns "ended" after saying goodbye and stopping the session.
        """
        txt = (user_text or "").strip().lower()

        # simple keyword match (can replace with regex or intent classifier)
        if any(k in txt for k in GOODBYE_KEYWORDS):
            # polite farewell
            farewell = f"Goodbye {self.meta.get('user_name', '')}. Ending the call now."
            await self.say(farewell)

            # stop the running session (session set by runtime)
            if getattr(self, "session", None) is not None:
                await self.session.stop()
            return json.dumps({"status": "ended", "farewell": farewell})
        return json.dumps({"status": "continue"})

2) Ensure AgentSession sets

agent.session

so the tool can call stop()

async def entrypoint(ctx: JobContext):
    meta_data = await get_all_metadata(ctx)  # dict from dispatch-time metadata

    # runtime values known at session start (populate from your runtime source)
    runtime = {
        "intent_name": "book_meeting",
        "intent_params": {"duration_minutes": 30},
        "user_name": "Arun",
        "email": "<mailto:arun@arun.com|arun@arun.com>",
        "timezone": "America/New_York",
    }

    merged = {**meta_data, **runtime}
    agent = V1TestAgent(merged)
    session = AgentSession()

    # attach session to agent so tools can stop it
    agent.session = session

    await session.start(agent=agent, room=ctx.room, context=runtime)
    await session.say(merged['agent'].get('greeting_message', WELCOME_MESSAGE))

    # keep entrypoint alive while session runs (if needed)
    await session.wait_closed()  # or equivalent wait for session termination

3) Where to call the tool • If you run user messages through tools automatically, call

check_goodbye_and_maybe_end

early for each message and short-circuit further handling when it returns

"ended"

. • If your agent pipeline uses an LLM step that chooses tools, the LLM can call

check_goodbye_and_maybe_end(user_text=...)

as a tool; the tool will stop the session when it detects a goodbye. Example usage in a message handler (pseudo):

# when a user message arrives
res = await agent.check_goodbye_and_maybe_end(user_text=incoming_text)
out = json.loads(res)
if out["status"] == "ended":
    return  # session already stopped
# else continue normal handling

Notes • Use

session.stop()

to gracefully end the call so your

on_session_end

cleanup runs. • Keyword matching is simple; swap in regex or an intent classifier if you need more accuracy. • Passing runtime metadata into

Agent

(as

self.meta

) lets the goodbye message be personalized. If you want, I can: • Swap the simple keyword check for a small regex-based detector, or • Show an example where the LLM is allowed to call the tool (tool signature + expected LLM tool call). Which would you prefer?

import json import httpx from dotenv import load_dotenv from livekit.agents import Agent, JobContext, JobProcess, WorkerOptions, cli, function_tool from livekit.agents.voice import AgentSession from livekit.plugins import silero from agents.utils import get_llm_from_config, get_stt_from_config, get_tts_from_config from agents.utils import WELCOME_MESSAGE, DEFAULT_SYSTEM_PROMPT from agents.utils.config import get_all_metadata from agents.functions.caldotcom import get_free_slots, book_slot load_dotenv() def prewarm(proc: JobProcess): proc.userdata.update({ "vad": silero.VAD.load(), }) class V1TestAgent(Agent): def __init__(self, meta_data): agent_info = meta_data['agent'] super().__init__( instructions=agent_info.get("system_prompt", DEFAULT_SYSTEM_PROMPT), stt=get_stt_from_config(meta_data['stt']), llm=get_llm_from_config(meta_data['llm']), tts=get_tts_from_config(meta_data['tts']) ) async def on_enter(self): new_instructions = f""" You are a helpful scheduling assistant. The current user is arun (email: arun@arun.com) in timezone America/New_York. Always ask the user to confirm date/time before booking. """ await self.update_instructions(new_instructions) @function_tool() async def list_event_types(self, start_time: str, end_time: str) -> str: """ Fetch available slots for a given date/time range. Parameters should be ISO 8601 strings, e.g., "2025-12-29T000000Z" """ slots = await get_free_slots(start_time, end_time) return json.dumps(slots, indent=2) @function_tool() async def book_available_slot(self, start_time_utc: str, name: str, email: str) -> str: """ timezone="America/New_York" Book a slot using Cal.com API. """ timezone="America/New_York" eventTypeId=4264613 booking = await book_slot(start_time_utc, name, email, timezone, eventTypeId) if booking is None: return json.dumps({"error": "Failed to book slot"}, indent=2) return json.dumps(booking, indent=2) @function_tool() async def end_call_from_agent(self) -> str: """ Call this fucntion only when user say bye or willing to disconnect """ self.session.shutdown() await self.session.aclose() return json.dumps({"status": "session shutdown initiated"}, indent=2) async def entrypoint(ctx: JobContext): meta_data = await get_all_metadata(ctx) agent = V1TestAgent(meta_data) session = AgentSession() await session.start(agent=agent, room=ctx.room) if True: await session.say(meta_data['agent'].get('greeting_message', WELCOME_MESSAGE)) async def on_session_end(): print("ending") ctx.delete_room(ctx.room.name) ctx.add_shutdown_callback(on_session_end) def run(): cli.run_app( WorkerOptions( entrypoint_fnc=entrypoint, prewarm_fnc=prewarm, agent_name="v1_test_agent", port=8082, ) ) if name == "__main__": run() sesion not ending wi say to end async def on_session_end(): print("ending") ctx.delete_room(ctx.room.name) not claling

Thanks — I see the issue: your tool calls

self.session.shutdown()

/

aclose()

(and those methods may not trigger the normal shutdown flow), and your session object may not be attached to the agent. Use the AgentSession's async stop/close method and ensure the session is attached to the agent so the tool can call it. Also await the session stop so the entrypoint can continue and the shutdown callback runs. Minimal fixes — apply these changes: 1) Attach the session to the agent in entrypoint and wait for session close:

async def entrypoint(ctx: JobContext):
    meta_data = await get_all_metadata(ctx)
    agent = V1TestAgent(meta_data)
    session = AgentSession()

    # attach so tools can call stop()
    agent.session = session

    await session.start(agent=agent, room=ctx.room, context={})
    await session.say(meta_data['agent'].get('greeting_message', WELCOME_MESSAGE))

    # wait until session stops so on_session_end runs
    await session.wait_closed()

2) Use an async stop/close from the tool (await it). Replace your tool with:

@function_tool()
async def end_call_from_agent(self) -> str:
    # polite farewell before stopping
    await self.say("Goodbye — ending the call now.")
    # ensure session exists
    if getattr(self, "session", None) is None:
        return json.dumps({"error": "no session attached"}, indent=2)

    # use the session's async stop/close API and await it
    await self.session.stop()        # preferred graceful stop
    # or await self.session.close() depending on API
    return json.dumps({"status": "session stop initiated"}, indent=2)

3) Ensure your shutdown callback is registered before session runs (you already add it), and that it performs final cleanup:

async def on_session_end():
    print("ending")
    ctx.delete_room(ctx.room.name)

ctx.add_shutdown_callback(on_session_end)

4) Where to call the tool • If the LLM calls tools, the LLM can call `end_call_from_agent()`; the tool will stop the session. • If you call the tool from a message handler, call it and then return/stop further processing after it returns. Debug tips if on_session_end still doesn't run: • Confirm

agent.session

is the same session returned by

AgentSession()

(print or log id). • Ensure

await session.stop()

or

await session.wait_closed()

is reached (add logs before/after). • If your runtime uses a different stop method name, use the session's async graceful stop (look for

stop()

,

close()

, or

wait_closed()

on AgentSession). If you want, I can: • Provide a single full-file corrected example wired to your current code, or • Change the goodbye detection to be called automatically each incoming message (tool-first flow). Which would you prefer?

"function": "end_call_from_agent", "arguments": "{}", "speech_id": "speech_7eace3ac59ed"} 151844.279 ERROR livekit.agents exception occurred while executing tool {"function": "end_call_from_agent", "speech_id": "speech_7eace3ac59ed"} Traceback (most recent call last): File "C:\Users\work\work\agents\.venv\Lib\site-packages\livekit\agents\voice\generation.py", line 597, in _traceable_fnc_tool val = await function_callable() ^^^^^^^^^^^^^^^^^^^^^^^^^ File "C:\Users\work\work\agents\agents\testagents\web\v1_test_agent.py", line 65, in end_call_from_agent await self.say("Goodbye — ending the call now.") ^^^^^^^^ AttributeError: 'V1TestAgent' object has no attribute 'say' 151844.291 DEBUG livekit.agents tools execution completed {"speech_id": "speech_7eace3ac59ed"} 151844.994 DEBUG grpc._cy…on.cygrpc [_cygrpc] Loaded running loop: id(loop)=2615845648240 Headset Microphone (Jabra EVOLV ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ ▁ <Ctrl+T> Text Mode

Thanks — that AttributeError explains why the tool failed. The Agent instance doesn't have a

say()

helper; the session does. Do these minimal fixes: 1) Attach the session object to the agent in entrypoint:

async def entrypoint(ctx: JobContext):
    meta_data = await get_all_metadata(ctx)
    agent = V1TestAgent(meta_data)
    session = AgentSession()
    agent.session = session            # <--- attach here
    await session.start(agent=agent, room=ctx.room, context={})
    await session.say(meta_data['agent'].get('greeting_message', WELCOME_MESSAGE))
    await session.wait_closed()

2) Update the tool to call session methods (await them) instead of `self.say`:

@function_tool()
async def end_call_from_agent(self) -> str:
    if getattr(self, "session", None) is None:
        return json.dumps({"error": "no session attached"}, indent=2)

    # polite farewell via session
    await self.session.say("Goodbye — ending the call now.")
    # graceful stop and wait for close so shutdown callbacks run
    await self.session.stop()
    await self.session.wait_closed()
    return json.dumps({"status": "session stopped"}, indent=2)

3) Ensure your shutdown callback is registered before starting session:

async def on_session_end():
    print("ending")
    ctx.delete_room(ctx.room.name)

ctx.add_shutdown_callback(on_session_end)

4) Debugging tips if it still doesn't trigger: • Log/print

agent.session

in the tool to confirm it's the same object created in entrypoint. • Add logs before/after

await self.session.stop()

and in

on_session_end

to confirm flow. • If your AgentSession API uses different method names, replace

stop()

/

wait_closed()

with the session's graceful stop/close APIs. If you want, I can produce a single full-file corrected example integrating these changes. Would you like that?

can u geve me best tts with low lentecncy and realtim tts with livekit agents

Nice — here are the best low‑latency, real‑time TTS options you can use with LiveKit agents and when to pick each. Tell me which language/voice style and which SDK (Python or Node) you’re using and I’ll give a short setup snippet. • Cartesia (recommended) ◦ Lowest latency, built for streaming real‑time apps (models like

sonic-3

). ◦ Great when time-to-first-byte matters. ◦ Use when you need very fast, natural streaming voice. • OpenAI Realtime (best effective latency) ◦ Generates audio directly in the realtime multimodal flow (avoids separate TTS step). ◦ Best if you can use a RealtimeModel/MultimodalAgent for audio-to-audio interaction. • Groq ◦ Low-latency inference on specialized hardware; good if you already use Groq for LLMs. ◦ Use when you need ultra-fast inference and have Groq infra. • ElevenLabs ◦ Very natural, expressive voices with streaming support. ◦ Slightly higher latency than Cartesia/OpenAI RT but better voice quality/control. • Rime ◦ Good balance of responsiveness and expressiveness for realtime agents. Practical tips • For lowest perceptible latency prefer Cartesia or OpenAI Realtime. • If you need highest naturalness with acceptable latency, use ElevenLabs. • Stream audio output into the AgentSession’s

tts

pipeline so audio plays as it’s generated. • Personalize voices and prosody (speed/emotion) where the provider supports it to mask small latencies. • Test time-to-first-byte and time-to-speech in your real environment (network + client playback matter). If you want, I’ll provide: • A minimal Python snippet wiring Cartesia

sonic-3

into your existing AgentSession, or • A short example using OpenAI Realtime multimodal agent (audio pipeline) for the lowest end‑to‑end latency. Which example would you like? Sources: Cartesia TTS plugin guide | LiveKit docs | ElevenLabs TTS plugin guide | LiveKit docs | OpenAI Realtime API plugin guide | LiveKit docs | Groq TTS plugin guide | LiveKit docs | Text-to-speech (TTS) models | LiveKit Docs

• Stream audio output into the AgentSession’s

tts

pipeline so audio plays as it’s generated. how

Great — below is a minimal, practical guide and Python example showing how to stream TTS audio into an AgentSession so audio plays as it’s generated. I’ll use Cartesia (

sonic-3

) as the low-latency example; the same pattern applies for other streaming TTS providers (OpenAI Realtime, ElevenLabs) supported by LiveKit. Steps (high level) 1. Import the TTS plugin for your provider (Cartesia in this example). 2. Create the AgentSession with the TTS instance passed in. 3. Start the session — AgentSession will stream generated audio into the LiveKit room as TTS is produced. 4. Trigger speech via

session.say()

or have the agent produce replies that use the session TTS pipeline. Minimal Python example (Cartesia sonic-3)

from livekit.agents.voice import AgentSession
from livekit.plugins.cartesia import TTS as CartesiaTTS
# other imports omitted for brevity

# create TTS instance with low-latency streaming model
tts = CartesiaTTS(model="sonic-3", voice="default_voice_id")

# create session with the TTS attached
session = AgentSession(
    # pass your agent, stt, llm elsewhere when starting; showing only tts wiring here
    tts=tts
)

# start session (inside your entrypoint)
await session.start(agent=agent, room=ctx.room, context=runtime_context)

# when you want to speak, either:
await session.say("Hi Arun — I found available slots on Dec 30 at 3pm. Do you want me to book?")

# OR ensure your agent's LLM replies are routed through session's tts pipeline;
# session will stream audio as the TTS generates it.

Key points and best practices • AgentSession handles sending generated audio to the LiveKit room; you only need to attach a streaming-capable

tts

instance on creation. • Use providers/models labeled for streaming/real-time (Cartesia

sonic-3

, ElevenLabs streaming config, OpenAI Realtime multimodal) to minimize time-to-first-byte. • If provider requires credentials or streaming options, initialize the plugin accordingly (provide API key, voice id, or set streaming flags in the plugin constructor). • For polite UX: send a short initial prompt or SSML that elicits short first audio and then continue streaming longer content to reduce perceived latency. • Test TTFB and playback on your target client — client buffering and network conditions affect perceived latency. Use small wave chunk sizes if configurable. Alternatives • OpenAI Realtime (use Multimodal/RealtimeModel) can bypass separate TTS by generating audio directly; attach the realtime model instead of a separate TTS plugin. • ElevenLabs supports streaming voices; instantiate its TTS plugin and pass to

AgentSession

the same way. If you want, I can give the exact Cartesia plugin constructor with authentication, or a Node example for the same setup — which one do you prefer?

if i send stt then llm take time to resond now hoe cna we imporve that

You're asking how to reduce latency between STT and LLM response in your LiveKit agent. Below are focused, practical ways to speed up the STT → LLM → TTS pipeline and quick actionable items you can try now. High‑impact options (try in this order) • Use realtime audio models that avoid a separate STT+TTS roundtrip: ◦ OpenAI Realtime / MultimodalModel (audio‑to‑audio) — generates audio directly, removing STT → text → TTS delay. • Use streaming models and token‑level streaming: ◦ Enable LLM streaming so you get partial tokens as they’re generated instead of waiting for full completion. ◦ Configure your LLM client/session to stream and play audio while tokens arrive. • Use low‑latency TTS that streams (Cartesia, ElevenLabs streaming, Rime): ◦ Attach streaming TTS to AgentSession so TTS audio starts before full text is complete. STT optimizations • Use a fast streaming STT (Silero or other low-latency models) with partial interim transcripts so you can start processing before final transcript arrives. • Send interim partial transcripts to the LLM (as prefix) to start response generation early. LLM optimizations • Use smaller, faster LLMs for real‑time responses (e.g., gpt-4o-mini, or other low-latency options) for conversational turns; reserve larger models for heavy tasks. • Stream tokens from the LLM and start TTS as soon as first tokens arrive. • Shorten prompts and keep system/context minimal; avoid re-sending long histories each turn (use summarized context). Parallel & architectural improvements • Pipeline partial STT → LLM streaming → TTS streaming concurrently rather than sequentially. • Prewarm LLM/TTS instances and keep warm processes (avoid cold starts). • Run LLMs/TTS close to your LiveKit servers (same region) and use GPU instances if available. • Cache frequent prompts/responses and reuse results when possible. Prompt & model tuning • Use concise system prompt and task-specific instructions to reduce generation time. • Limit max_tokens and use temperature/penalty settings appropriate for speed. • If you use tools/functions, avoid unnecessary tool calls or make them synchronous only when needed. Concrete pattern (minimal flow) 1. Start streaming STT with interim partials. 2. Send partials immediately to the LLM stream endpoint (or RealtimeModel). 3. Start streaming TTS as soon as LLM yields tokens (pipe tokens into TTS streaming). 4. Continue streaming until LLM finalizes. Minimal code notes (conceptual) • Use AgentSession with streaming-enabled LLM and streaming TTS (Cartesia or OpenAI Realtime). • Enable LLM streaming in your LLM client and hook token/callback to push to TTS: ◦ On new LLM token -> append to buffer -> call session.tts.stream_chunk(token_chunk) / session.say for accumulated chunk. • Use interim STT events to feed LLM as they arrive. Next step I can provide • A short Python example wiring: Silero streaming STT (with interim), a streaming LLM client (gpt-4o-mini or RealtimeModel) with token callbacks, and Cartesia TTS attached to AgentSession so audio plays as tokens arrive. Tell me whether you want: Python or Node, and which TTS/LLM provider you prefer.

import json from datetime import datetime from zoneinfo import ZoneInfo from dotenv import load_dotenv from livekit.agents import Agent, JobContext, JobProcess, WorkerOptions, cli, function_tool from livekit.agents.voice import AgentSession from livekit.plugins import silero from agents.functions.caldotcom import tools from agents.utils import get_llm_from_config, get_stt_from_config, get_tts_from_config from agents.utils import WELCOME_MESSAGE, DEFAULT_SYSTEM_PROMPT from agents.utils.config import get_all_metadata load_dotenv() def prewarm(proc: JobProcess): proc.userdata.update({ "vad": silero.VAD.load(), }) class V1TestAgent(Agent): def __init__(self, ctx, meta_data): self.ctx = ctx agent_info = meta_data['agent'] super().__init__( instructions=agent_info.get("system_prompt", DEFAULT_SYSTEM_PROMPT), stt=get_stt_from_config(meta_data['stt']), llm=get_llm_from_config(meta_data['llm']), tts=get_tts_from_config(meta_data['tts']), tools=tools, ) async def on_enter(self): now = datetime.now(ZoneInfo("America/New_York")) current_time = now.strftime("%Y-%m-%d %H:%M %Z") new_instructions = f""" You are a helpful scheduling assistant. Current date and time: {current_time} The user is Arun (email: arun@arun.com), timezone America/New_York. Always ask the user to confirm date/time before booking. """ await self.update_instructions(new_instructions) @function_tool() async def end_call_on_user_request(self) -> str: """ Call this fucntion only when user say bye or willing to disconnect """ self.ctx.shutdown() return json.dumps({"status": "session shutdown initiated"}, indent=2) async def entrypoint(ctx: JobContext): meta_data = await get_all_metadata(ctx) agent = V1TestAgent(ctx, meta_data) session = AgentSession() await session.start(agent=agent, room=ctx.room) if True: await session.say(meta_data['agent'].get('greeting_message', WELCOME_MESSAGE)) async def on_session_end(): ctx.delete_room(ctx.room.name) ctx.add_shutdown_callback(on_session_end) def run(): cli.run_app( WorkerOptions( entrypoint_fnc=entrypoint, prewarm_fnc=prewarm, agent_name="v1_test_agent", port=8082, ) ) if name == "__main__": run() reduce and apply all above i confgs

Thanks — I’ll give a concise, corrected Python example that integrates your requirements: runtime data injection, streaming TTS into AgentSession, low‑latency optimizations (preemptive/streaming), and a safe function tool to end the session. Apply these minimal edits to your code. Key changes • Configure STT/LLM/TTS on AgentSession (session manages pipeline). • Inject runtime data into agent instructions at creation. • Attach

session

to the agent so tools can stop/shutdown the session. • Use streaming / preemptive generation options to reduce latency. • Use an awaited graceful shutdown inside the tool so shutdown callbacks run. Minimal integrated example

import json
from datetime import datetime
from zoneinfo import ZoneInfo
from dotenv import load_dotenv
from livekit.agents import Agent, JobContext, JobProcess, WorkerOptions, cli, function_tool
from livekit.agents.voice import AgentSession
from livekit.plugins import silero
from agents.functions.caldotcom import tools
from agents.utils import get_llm_from_config, get_stt_from_config, get_tts_from_config
from agents.utils import WELCOME_MESSAGE, DEFAULT_SYSTEM_PROMPT
from agents.utils.config import get_all_metadata

load_dotenv()

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

class V1TestAgent(Agent):
    def __init__(self, meta_data):
        # build dynamic instructions once at creation
        agent_info = meta_data['agent']
        now = datetime.now(ZoneInfo(meta_data.get("timezone", "America/New_York")))
        current_time = now.strftime("%Y-%m-%d %H:%M %Z")
        instructions = (
            agent_info.get("system_prompt", DEFAULT_SYSTEM_PROMPT)
            + "\n\n"
            + f"Current date/time: {current_time}\n"
            + f"User: {meta_data.get('user_name','Unknown')} (email: {meta_data.get('email','')})\n"
            + "Always ask the user to confirm date/time before booking."
        )

        # Agent created without explicit stt/llm/tts (session supplies them)
        super().__init__(instructions=instructions, tools=tools)
        self.meta = meta_data
        self.session = None  # will be set by entrypoint

    @function_tool()
    async def end_call_on_user_request(self) -> str:
        # graceful shutdown: wait for session to finish speaking and close
        if getattr(self, "session", None) is None:
            return json.dumps({"error": "no session attached"}, indent=2)
        await self.session.say(f"Goodbye {self.meta.get('user_name','')}. Ending the call now.")
        await self.session.shutdown(drain=True)  # graceful drain
        await self.session.wait_closed()
        return json.dumps({"status": "session shutdown initiated"}, indent=2)

async def entrypoint(ctx: JobContext):
    meta_data = await get_all_metadata(ctx)  # dispatch-time metadata dict

    # Create session with streaming/preemptive options for low latency
    session = AgentSession(
        stt=get_stt_from_config(meta_data['stt']),
        llm=get_llm_from_config(meta_data['llm']),
        tts=get_tts_from_config(meta_data['tts']),
        vad=ctx.proc.userdata.get("vad"),
        preemptive_generation=True,   # start LLM earlier for lower perceived latency
        # If your LLM supports token streaming, ensure it's enabled in get_llm_from_config
    )

    agent = V1TestAgent(meta_data)
    agent.session = session  # attach so tools can access session methods

    await session.start(agent=agent, room=ctx.room, context={
        "user_name": meta_data.get("user_name"),
        "timezone": meta_data.get("timezone"),
        "email": meta_data.get("email"),
        "intent_name": meta_data.get("intent_name"),
        "intent_params": meta_data.get("intent_params", {}),
    })

    # initial greeting
    await session.say(meta_data['agent'].get('greeting_message', WELCOME_MESSAGE))

    # ensure cleanup runs after session ends
    async def on_session_end():
        try:
            await ctx.api.room.delete_room({"room": ctx.room.name})
        except Exception:
            pass

    ctx.add_shutdown_callback(on_session_end)

    # wait until session finishes so entrypoint doesn't exit early
    await session.wait_closed()

def run():
    cli.run_app(WorkerOptions(
        entrypoint_fnc=entrypoint,
        prewarm_fnc=prewarm,
        agent_name="v1_test_agent",
        port=8082,
    ))

if __name__ == "__main__":
    run()

Practical notes and next steps • Ensure your

get_llm_from_config

enables token streaming (streaming callbacks) so you can begin TTS as soon as tokens arrive. Pipe tokens into streaming TTS when possible. • Use a streaming-capable TTS (Cartesia sonic-3, ElevenLabs streaming, or OpenAI Realtime) in

get_tts_from_config

. AgentSession will stream audio into the LiveKit room as TTS generates. •

preemptive_generation=True

reduces perceived latency by starting generation earlier from interim transcripts. Also use interim STT partials if your STT supports them. • Debug tips: add logs before/after

await session.shutdown(...)

, and log

agent.session is not None

to confirm attachment. If shutdown still doesn't run, ensure the tool gets called by the LLM (check tool call logs) and that there are no uncaught exceptions. If you want, I can: • Provide the exact Cartesia or OpenAI Realtime constructor to use in

get_tts_from_config

, or • Add a token-stream callback example that feeds tokens into TTS for immediate audio playback. Which one should I give you next? Sources: Job lifecycle | LiveKit docs | examples/voice_agents | livekit/livekit_composite | DeepWiki | Simple Call Answering Agent | LiveKit docs | Agent speech and audio | LiveKit docs | Collect recording consent with tasks | LiveKit docs

Using proactor: IocpProactor 180248.479 INFO livekit.agents starting worker {"version": "1.3.10", "rtc-version": "1.0.23"} 180248.482 INFO livekit.agents HTTP server listening on :8082 180248.507 INFO livekit.agents initializing job runner {"tid": 19792} 180248.697 INFO livekit.agents job runner initialized {"tid": 19792, "elapsed_time": 0.19} DEBUG asyncio Using proactor: IocpProactor 180248.839 ERROR livekit.agents unhandled exception while running the job task Traceback (most recent call last): File "C:\Users\work\work\agents\.venv\Lib\site-packages\opentelemetry\util\_decorator.py", line 71, in async_wrapper return await func(*args, **kwargs) # type: ignore ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "C:\Users\work\work\agents\.venv\Lib\site-packages\livekit\agents\ipc\job_proc_lazy_main.py", line 286, in _traceable_entrypoint await self._job_entrypoint_fnc(job_ctx) File "C:\Users\work\work\agents\agents\testagents\web\v1_test_agent.py", line 60, in entrypoint agent = V1TestAgent(meta_data) ^^^^^^^^^^^^^^^^^^^^^^ File "C:\Users\work\work\agents\agents\testagents\web\v1_test_agent.py", line 35, in init self.session = None # will be set by entrypoint ^^^^^^^^^^^^ AttributeError: property 'session' of 'V1TestAgent' object has no setter

Thanks — that AttributeError means

Agent

defines a read‑only

session

property, so assigning

self.session = ...

fails. Fix by using a different attribute name (e.g.,

self._session

) and reference that in your tool. Minimal fixes: 1) In your agent class, use

_session

instead of `session`:

class V1TestAgent(Agent):
    def __init__(self, meta_data):
        super().__init__(instructions=..., tools=tools)
        self.meta = meta_data
        self._session = None

2) Attach session in entrypoint to the private attribute:

agent = V1TestAgent(meta_data)
agent._session = session

3) Update your end tool to use

_session

and await graceful shutdown:

@function_tool()
async def end_call_on_user_request(self) -> str:
    if getattr(self, "_session", None) is None:
        return json.dumps({"error": "no session attached"}, indent=2)
    await self._session.say(f"Goodbye {self.meta.get('user_name','')}.")
    await self._session.shutdown(drain=True)
    await self._session.wait_closed()
    return json.dumps({"status": "session shutdown initiated"}, indent=2)

4) Add logs to verify attachment and flow: • Log

agent._session is None

right after attaching. • Log before/after shutdown in the tool and in

on_session_end

. That will remove the AttributeError and allow the tool to shut down the session so your shutdown callback runs. Want me to produce a single updated file with these changes applied?

fukll code

Below is a single, full Python file that integrates the fixes and features we discussed: • dynamic runtime data injected into agent instructions • STT / LLM / TTS configured on AgentSession (streaming-capable) • low-latency options: preemptive_generation and token streaming (ensure your get_*_from_config enable streaming) • graceful session termination via a function tool (uses a private

_session

attribute to avoid read-only property) • waits for session to close so shutdown callbacks run and room is deleted Update any provider-specific constructors inside your get_*_from_config implementations (they must return streaming-capable instances). Save as e.g. v1_test_agent_full.py and run.

import json
import os
from datetime import datetime
from zoneinfo import ZoneInfo
from dotenv import load_dotenv

from livekit import api as lk_api
from livekit.agents import Agent, JobContext, JobProcess, WorkerOptions, cli, function_tool
from livekit.agents.voice import AgentSession
from livekit.plugins import silero

# Your project-specific imports - ensure these exist and return streaming-capable objects
from agents.functions.caldotcom import tools
from agents.utils import get_llm_from_config, get_stt_from_config, get_tts_from_config
from agents.utils import WELCOME_MESSAGE, DEFAULT_SYSTEM_PROMPT
from agents.utils.config import get_all_metadata

load_dotenv()

def prewarm(proc: JobProcess):
    # Preload VAD once per process for faster startup
    proc.userdata["vad"] = silero.VAD.load()

class V1TestAgent(Agent):
    """
    Scheduling agent example. AgentSession provides stt/llm/tts; we attach it to the agent
    using a private attribute `_session` so we don't write to a read-only property.
    """
    def __init__(self, meta_data: dict):
        # Build dynamic instructions using runtime metadata
        agent_info = meta_data.get('agent', {})
        timezone = meta_data.get("timezone", "America/New_York")
        now = datetime.now(ZoneInfo(timezone))
        current_time = now.strftime("%Y-%m-%d %H:%M %Z")

        instructions = (
            agent_info.get("system_prompt", DEFAULT_SYSTEM_PROMPT)
            + "\n\n"
            + f"Current date/time: {current_time}\n"
            + f"User: {meta_data.get('user_name','Unknown')} (email: {meta_data.get('email','')})\n"
            + "Always ask the user to confirm date/time before booking."
        )

        # Initialize Agent with instructions and tools. STT/LLM/TTS live on the session.
        super().__init__(instructions=instructions, tools=tools)
        self.meta = meta_data
        self._session = None  # attach the AgentSession instance here from entrypoint

    @function_tool()
    async def list_event_types(self, start_time: str, end_time: str) -> str:
        slots = await tools.get_free_slots(start_time, end_time)
        return json.dumps(slots, indent=2)

    @function_tool()
    async def book_available_slot(self, start_time_utc: str, name: str, email: str) -> str:
        timezone = self.meta.get("timezone", "America/New_York")
        eventTypeId = 4264613
        booking = await tools.book_slot(start_time_utc, name, email, timezone, eventTypeId)
        if booking is None:
            return json.dumps({"error": "Failed to book slot"}, indent=2)
        return json.dumps(booking, indent=2)

    @function_tool()
    async def end_call_on_user_request(self) -> str:
        """
        Gracefully end the session when the user requests to disconnect.
        Uses the attached `_session` for polite farewell and shutdown.
        """
        if getattr(self, "_session", None) is None:
            return json.dumps({"error": "no session attached"}, indent=2)

        # polite farewell
        try:
            await self._session.say(f"Goodbye {self.meta.get('user_name','')}. Ending the call now.")
        except Exception:
            # ignore TTS errors for farewell
            pass

        # graceful shutdown: drain ensures in-progress audio finishes
        await self._session.shutdown(drain=True)
        # wait for full close
        await self._session.wait_closed()
        return json.dumps({"status": "session shutdown initiated"}, indent=2)

async def entrypoint(ctx: JobContext):
    # Load dispatch-time metadata (should return a dictionary)
    meta_data = await get_all_metadata(ctx)

    # Build session with streaming-capable components
    session = AgentSession(
        stt=get_stt_from_config(meta_data['stt']),
        llm=get_llm_from_config(meta_data['llm']),
        tts=get_tts_from_config(meta_data['tts']),
        vad=ctx.proc.userdata.get("vad"),
        preemptive_generation=True,  # start LLM earlier using interim STT
        # If get_llm_from_config supports token streaming, ensure it's enabled there
    )

    agent = V1TestAgent(meta_data)
    agent._session = session  # attach session to agent using private attr

    # Start the session (streams STT -> LLM -> TTS)
    await session.start(agent=agent, room=ctx.room, context={
        "user_name": meta_data.get("user_name"),
        "timezone": meta_data.get("timezone"),
        "email": meta_data.get("email"),
        "intent_name": meta_data.get("intent_name"),
        "intent_params": meta_data.get("intent_params", {}),
    })

    # Send initial greeting
    await session.say(meta_data['agent'].get('greeting_message', WELCOME_MESSAGE))

    # Register shutdown callback to delete the room when job ends
    async def on_session_end():
        try:
            # Use the LiveKit API client to delete the room
            api_client = lk_api.LiveKitAPI(
                os.getenv("LIVEKIT_URL"),
                os.getenv("LIVEKIT_API_KEY"),
                os.getenv("LIVEKIT_API_SECRET"),
            )
            await api_client.room.delete_room({"room": ctx.room.name})
        except Exception:
            pass

    ctx.add_shutdown_callback(on_session_end)

    # Wait until session closes so entrypoint doesn't exit prematurely
    await session.wait_closed()

def run():
    cli.run_app(WorkerOptions(
        entrypoint_fnc=entrypoint,
        prewarm_fnc=prewarm,
        agent_name="v1_test_agent",
        port=int(os.getenv("PORT", "8082")),
    ))

if __name__ == "__main__":
    run()

Notes and checklist to make this work in your environment • get_stt_from_config / get_llm_from_config / get_tts_from_config must return streaming-capable objects and enable token streaming/preemptive behavior if supported. • If you use Cartesia/ElevenLabs/OpenAI Realtime TTS, ensure API keys and voice IDs are supplied inside get_tts_from_config. • Confirm your tools import paths (I used tools.get_free_slots and tools.book_slot earlier; adapt to your module layout). • Add logging around critical calls (session.start, session.shutdown) if you need deeper debugging. • If you need an example get_tts_from_config for Cartesia or OpenAI Realtime, tell me which provider and I’ll provide that constructor snippet next.

Microsoft Windows [Version 10.0.26200.7462] (c) Microsoft Corporation. All rights reserved. (agents) C\Users\work\work\agents&gt;c/Users/work/work/agents/.venv/Scripts/activate.bat (agents) C:\Users\work\work\agents> (agents) C:\Users\work\work\agents> uv run python -m agents.testagents.web.v1_test_agent console Agents Starting console mode 🚀 180516.162 DEBUG asyncio Using proactor: IocpProactor 180516.166 INFO livekit.agents starting worker {"version": "1.3.10", "rtc-version": "1.0.23"} 180516.168 INFO livekit.agents HTTP server listening on :8082 180516.186 INFO livekit.agents initializing job runner {"tid": 31012} 180516.313 DEBUG asyncio Using proactor: IocpProactor 180516.314 INFO livekit.agents job runner initialized {"tid": 31012, "elapsed_time": 0.13} 180516.403 ERROR livekit.agents unhandled exception while running the job task Traceback (most recent call last): File "C:\Users\work\work\agents\.venv\Lib\site-packages\opentelemetry\util\_decorator.py", line 71, in async_wrapper return await func(*args, **kwargs) # type: ignore ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "C:\Users\work\work\agents\.venv\Lib\site-packages\livekit\agents\ipc\job_proc_lazy_main.py", line 286, in _traceable_entrypoint await self._job_entrypoint_fnc(job_ctx) File "C:\Users\work\work\agents\agents\testagents\web\v1_test_agent.py", line 66, in entrypoint await session.start(agent=agent, room=ctx.room, context={ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ TypeError: AgentSession.start() got an unexpected keyword argument 'context'

Thanks — the error is because

AgentSession.start()

does not accept a

context

parameter. Remove

context=...

and pass runtime data into the Agent (or ChatContext) before starting the session. Below is a corrected minimal file you can run — it removes the unsupported

context

argument, attaches the session to the agent via

_session

, and keeps graceful shutdown and streaming options. Save and run as your agent module (adjust get_*_from_config functions and provider keys as needed):

import json
import os
from datetime import datetime
from zoneinfo import ZoneInfo
from dotenv import load_dotenv

from livekit import api as lk_api
from livekit.agents import Agent, JobContext, JobProcess, WorkerOptions, cli, function_tool
from livekit.agents.voice import AgentSession
from livekit.plugins import silero

from agents.functions.caldotcom import tools
from agents.utils import get_llm_from_config, get_stt_from_config, get_tts_from_config
from agents.utils import WELCOME_MESSAGE, DEFAULT_SYSTEM_PROMPT
from agents.utils.config import get_all_metadata

load_dotenv()

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

class V1TestAgent(Agent):
    def __init__(self, meta_data: dict):
        agent_info = meta_data.get('agent', {})
        timezone = meta_data.get("timezone", "America/New_York")
        now = datetime.now(ZoneInfo(timezone))
        current_time = now.strftime("%Y-%m-%d %H:%M %Z")

        instructions = (
            agent_info.get("system_prompt", DEFAULT_SYSTEM_PROMPT)
            + "\n\n"
            + f"Current date/time: {current_time}\n"
            + f"User: {meta_data.get('user_name','Unknown')} (email: {meta_data.get('email','')})\n"
            + "Always ask the user to confirm date/time before booking."
        )

        super().__init__(instructions=instructions, tools=tools)
        self.meta = meta_data
        self._session = None  # attach AgentSession from entrypoint

    @function_tool()
    async def end_call_on_user_request(self) -> str:
        if getattr(self, "_session", None) is None:
            return json.dumps({"error": "no session attached"}, indent=2)
        try:
            await self._session.say(f"Goodbye {self.meta.get('user_name','')}. Ending the call now.")
        except Exception:
            pass
        await self._session.shutdown(drain=True)
        await self._session.wait_closed()
        return json.dumps({"status": "session shutdown initiated"}, indent=2)

async def entrypoint(ctx: JobContext):
    meta_data = await get_all_metadata(ctx)

    session = AgentSession(
        stt=get_stt_from_config(meta_data['stt']),
        llm=get_llm_from_config(meta_data['llm']),
        tts=get_tts_from_config(meta_data['tts']),
        vad=ctx.proc.userdata.get("vad"),
        preemptive_generation=True,
    )

    agent = V1TestAgent(meta_data)
    agent._session = session

    # DO NOT pass context=... to session.start()
    await session.start(agent=agent, room=ctx.room)

    # initial greeting (pass runtime data via agent.meta)
    await session.say(meta_data['agent'].get('greeting_message', WELCOME_MESSAGE))

    async def on_session_end():
        try:
            api_client = lk_api.LiveKitAPI(
                os.getenv("LIVEKIT_URL"),
                os.getenv("LIVEKIT_API_KEY"),
                os.getenv("LIVEKIT_API_SECRET"),
            )
            await api_client.room.delete_room({"room": ctx.room.name})
        except Exception:
            pass

    ctx.add_shutdown_callback(on_session_end)

    await session.wait_closed()

def run():
    cli.run_app(WorkerOptions(
        entrypoint_fnc=entrypoint,
        prewarm_fnc=prewarm,
        agent_name="v1_test_agent",
        port=int(os.getenv("PORT", "8082")),
    ))

if __name__ == "__main__":
    run()

If you want, I can: • Provide the exact Cartesia or OpenAI Realtime TTS constructor to use inside

get_tts_from_config

, or • Show how to feed interim STT partials into the LLM for token streaming to further reduce latency. Which would you like next? Sources: Agent session | LiveKit docs | External data and RAG | LiveKit docs | Job lifecycle | LiveKit docs | Tool definition and use | LiveKit docs