In my AgentSession a active task in agent is running. I want to stop the agent from the agent session gracefuly and update the new_agent

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

I'm using python framework, I want to update the agent when new participant named "new_paticipant" joined in a session

Thanks — I understand you’re using the Python framework and you want the running AgentSession to gracefully stop the current agent/task and switch to a new agent when a participant with identity "new_participant" joins. Below is a concise, practical pattern you can use. Important idea • The running task must cooperate: it should check a cancellation flag or let

session.update_agent()

perform a graceful handoff. You’ll listen for the participant join event on the room and then call

session.update_agent(new_agent)

to perform the switch. Example (Python) • This example shows: 1) starting an AgentSession with an initial agent, 2) listening for participant joins, 3) switching to

NewAgent()

when a participant with identity "new_participant" joins, 4) notifying the session (optional).

import asyncio
from livekit_agents import AgentSession  # adjust import to your package layout
# import your agent classes
# from agents import InitialAgent, NewAgent

@server.rtc_session()
async def entrypoint(ctx: JobContext):
    session = AgentSession(
        # supply your configured stt, llm, tts, vad, etc.
    )

    initial_agent = InitialAgent()
    await session.start(agent=initial_agent, room=ctx.room)
    await ctx.connect()

    @ctx.room.on("participant_connected")
    def on_participant_connected(participant: rtc.RemoteParticipant):
        # check identity exactly as joined
        if participant.identity == "new_participant":
            # instantiate the new agent you want to switch to
            new_agent = NewAgent()

            # schedule the agent update (avoid blocking the event handler)
            asyncio.create_task(handle_handoff(session, new_agent))

async def handle_handoff(session: AgentSession, new_agent):
    # Option A: direct update_agent() — performs a graceful handoff
    await session.update_agent(new_agent)

    # Optionally, generate a short notification message from the new agent
    await session.generate_reply(instructions="Welcome new participant! Switching agents now.")

Notes and best practices • Use

asyncio.create_task()

in the event handler so you don’t block the room event loop. • If your current task runs long operations, ensure it can be interrupted or checks any session-level cancellation flag so cleanup happens promptly. You can store a flag on session userdata (for example

session.userdata["task_should_stop"] = True

) and have the task check it periodically. •

session.update_agent()

performs a graceful handoff: the session will stop the current agent/task and start the new agent so the session stays consistent. • Ensure

NewAgent

is a valid Agent subclass and ready to assume control of the session (tools, permissions, etc.). If you want, I can: • Provide a variant that uses a cancellation flag inside the running

AgentTask

(showing where to poll and exit), • Or tailor the code to your specific Agent subclasses if you paste small snippets of your current

InitialAgent

and task logic. Sources: Agents and handoffs | LiveKit Documentation

I initally used llm , stt , tts in AgentSession. how to disable llm and tts when handoff

Thanks — I understand you want to switch agents when a new participant joins and make sure the new agent has LLM and TTS disabled while keeping STT (or other components) as needed. Below is a clear, working pattern you can use in Python. Key idea • When you create the new agent for the handoff, construct it with

llm=None

and

tts=None

. The session will adopt the new agent's components, effectively disabling LLM and TTS. Keep

stt

if you still want transcription. Example (adds to the earlier participant-joined flow)

import asyncio
from livekit_agents import AgentSession  # adjust import to your package layout
# from agents import InitialAgent, NewAgent  # your agent classes

@server.rtc_session()
async def entrypoint(ctx: JobContext):
    session = AgentSession(
        stt=deepgram.STT(),
        llm=openai.LLM(),
        tts=elevenlabs.TTS(),
        vad=silero.VAD.load(),
    )

    initial_agent = InitialAgent()
    await session.start(agent=initial_agent, room=ctx.room)
    await ctx.connect()

    @ctx.room.on("participant_connected")
    def on_participant_connected(participant: rtc.RemoteParticipant):
        if participant.identity == "new_participant":
            asyncio.create_task(handoff_disable_llm_tts(session))

async def handoff_disable_llm_tts(session: AgentSession):
    # Instantiate the new agent with LLM and TTS disabled.
    # Pass the session's STT if you still want transcription.
    new_agent = NewAgent(llm=None, tts=None, stt=session.stt)

    # Graceful handoff: update the session's active agent
    await session.update_agent(new_agent)

    # Optional: have the new agent produce a short reply (no TTS, only text)
    await session.generate_reply(instructions="Agent switched: LLM and TTS are disabled.")

Notes and best practices • Use

asyncio.create_task()

from the event handler to avoid blocking the room event loop. • If the current agent is running a long task, ensure that task checks for cancellation or cooperatively stops so the handoff is prompt. • If you want transcription only, pass

stt=session.stt

when creating the new agent; leaving it out disables STT as well. • Setting

llm=None

and

tts=None

is the practical way to disable those capabilities — there’s no separate disable method required. If you want, I can: • Show how to add a session cancellation flag and where to poll it inside a long-running

AgentTask

, or • Tailor the example to your exact

InitialAgent

and

NewAgent

constructors if you paste those snippets. Which would help most? Sources: Agents and handoffs | LiveKit Documentation

104527.973 ERROR claim_ve…ion_agent Failed to update agent: a coroutine was expected, got None {"room": "voice_assistant_room_9001", "pid": 155430, "job_id": "AJ_iEywgQEXsHEg", "room_id": "RM_jABEpj3peJVU"}

Thanks — I can help diagnose and fix the "a coroutine was expected, got None" error and show the correct pattern for switching agents (disabling LLM and TTS) when "new_participant" joins. What the error means (brief) • That error happens when an async operation was expected but Python received None instead. In this context it usually means either: ◦

session.update_agent(...)

was not awaited, or ◦ the value passed into

update_agent

(or returned from a tool) was None or not a valid agent/coroutine. Checklist you can verify quickly 1. Are you awaiting the update call? ◦ Correct:

await session.update_agent(new_agent)

◦ Incorrect:

session.update_agent(new_agent)

or

await session.update_agent()

if

update_agent

returned None 1. Is

new_agent

a valid Agent instance? ◦ Ensure your

NewAgent(...)

constructor returns an Agent object and not None. ◦ If you build the agent inside a function, confirm the function returns the instance. 1. If you trigger the handoff from an event handler, run the update in an async task so you can await it: ◦ Use

asyncio.create_task(handoff(...))

inside the synchronous event callback, and inside

handoff()

use

await session.update_agent(...)

. Working pattern (concise example) • This example assumes you already listen for participant joins and want to disable LLM and TTS on the new agent while keeping STT.

import asyncio

async def handoff_disable_llm_tts(session: AgentSession):
    # create a new agent with llm and tts disabled
    new_agent = NewAgent(llm=None, tts=None, stt=session.stt)
    # await the coroutine — this must be awaited
    await session.update_agent(new_agent)
    # optional: generate a text reply (no TTS)
    await session.generate_reply(instructions="Switched agents; LLM and TTS disabled.")

# in your room handler (synchronous callback)
@ctx.room.on("participant_connected")
def on_participant_connected(participant):
    if participant.identity == "new_participant":
        # schedule the async handoff so you can await inside it
        asyncio.create_task(handoff_disable_llm_tts(session))

Common pitfalls that cause your error • Calling

await session.update_agent(None)

because

NewAgent(...)

returned None under some condition. • Forgetting to

await session.update_agent(...)

inside your async code path. • Returning an agent from a tool incorrectly (e.g., returning

(agent, "msg")

vs the expected shape) such that the framework receives None where it expects a coroutine. If you’d like, I can check a small snippet of your actual code. Please paste: • The participant-connected handler, • The

handoff

function (if any), • The

NewAgent

constructor (or how you instantiate the new agent). With that I’ll point out the exact line causing the error and show the minimal change to fix it. Sources: Agents and handoffs | LiveKit Documentation | SIP Lifecycle Management Agent | LiveKit Documentation | Building an Automated IVR Menu Caller | LiveKit Documentation

class Transcriber(Agent): def __init__(self): super().__init__( instructions="not-needed", stt=azure.STT( speech_key=os.getenv("AZURE_SPEECH_KEY"), speech_region=os.getenv("AZURE_SPEECH_REGION"), ), ) async def on_user_turn_completed(self, chat_ctx: llm.ChatContext, new_message: llm.ChatMessage): user_transcript = new_message.text_content logger.info(new_message) logger.info(f" -> {user_transcript}") return # Explicitly return to complete the coroutine #raise StopResponse() this my new agent class

Thanks — this is helpful. Your

Transcriber

agent class looks fine: it sets

stt

in the constructor and your

on_user_turn_completed

returns explicitly, so it should not itself return None. The error "a coroutine was expected, got None" means the code that triggers the handoff attempted to await something that was None (or called an async API without awaiting a coroutine). The most likely causes in your flow are: • update_agent was not awaited, or • the value passed to

update_agent

was None because the agent constructor returned None (not the case here), or • you called

await session.update_agent(...)

from a synchronous callback (so the awaited expression evaluated to None in that path). Checklist and fixes to apply 1) Ensure you await update_agent in an async context • Correct (inside an async function): await session.update_agent(Transcriber()) • Incorrect: session.update_agent(Transcriber()) # not awaited 2) If you call the handoff from a synchronous event handler (e.g., the room event callback), run an async task and await inside it • In your participant-connected callback, do: asyncio.create_task(handoff(session)) • And in handoff: async def handoff(session): await session.update_agent(Transcriber()) await session.generate_reply(instructions="Switched to Transcriber (LLM/TTS disabled).") 3) Make sure Transcriber() is an Agent instance • Your class inherits Agent and its init returns nothing (correct). Constructing Transcriber() returns the instance, not None — so this is fine. 4) Disable LLM and TTS for the new agent • Construct the new agent with llm=None and tts=None if your Agent constructor accepts them. Example: transcriber = Transcriber(llm=None, tts=None, stt=session.stt) • If your Transcriber class does not accept those args right now, add them to its init signature and pass them up to super(). Minimal integrated example • Adds async handoff triggered on participant join and shows disabling LLM/TTS:

import asyncio

# your Transcriber class (adjust to accept llm/tts if needed)
class Transcriber(Agent):
    def __init__(self, llm=None, tts=None, stt=None):
        super().__init__(
            instructions="not-needed",
            stt=stt,
            llm=llm,
            tts=tts,
        )

    async def on_user_turn_completed(self, chat_ctx: llm.ChatContext, new_message: llm.ChatMessage):
        user_transcript = new_message.text_content
        <http://logger.info|logger.info>(new_message)
        <http://logger.info|logger.info>(f" -> {user_transcript}")
        return

# event handler (synchronous callback from the room)
@ctx.room.on("participant_connected")
def on_participant_connected(participant):
    if participant.identity == "new_participant":
        asyncio.create_task(handoff_disable_llm_tts(session))

# async handoff that is properly awaited
async def handoff_disable_llm_tts(session: AgentSession):
    # keep STT, disable LLM and TTS
    transcriber = Transcriber(llm=None, tts=None, stt=session.stt)
    await session.update_agent(transcriber)
    await session.generate_reply(instructions="Switched agents; LLM and TTS disabled.")

If you still see the error after making these changes, please paste the exact snippet where you trigger the handoff (the participant callback and the handoff function). I’ll point to the exact line causing the coroutine/None mismatch. Sources: Agents and handoffs | LiveKit Documentation

Traceback (most recent call last): File "/home/azureuser/AI-development/entrypoint.py", line 145, in handoff_disable_llm_tts await session.update_agent(new_agent) TypeError: object NoneType can't be used in 'await' expression

Thanks — the traceback makes the root cause clear:

new_agent

is None when you call

await session.update_agent(new_agent)

. That’s why Python raises

TypeError: object NoneType can't be used in 'await' expression

. What to check and fix (quick checklist) • Verify how you create

new_agent

inside

handoff_disable_llm_tts

. It must be an Agent instance (e.g.,

Transcriber(...)

), not

None

. • If your Transcriber constructor currently doesn't accept llm/tts args, ensure you construct it with those values (or modify the constructor): ◦ Example:

transcriber = Transcriber(llm=None, tts=None, stt=session.stt)

• Add a guard before awaiting to catch None early:

if new_agent is None:
      logger.error("handoff failed: new_agent is None")
      return
  await session.update_agent(new_agent)

• Always await

update_agent

from an async function. If you call it from a synchronous room callback, schedule an async task and do the await there (you already do this pattern). Common places this becomes None • A conditional branch that returns None instead of an Agent. • A factory function that accidentally returns nothing. • Mistakenly assigning

new_agent = await some_async_constructor()

where the async function returns None. Minimal safe pattern you can paste into your code

async def handoff_disable_llm_tts(session: AgentSession):
    # Ensure Transcriber accepts llm/tts/stt in __init__
    new_agent = Transcriber(llm=None, tts=None, stt=session.stt)

    if new_agent is None:
        logger.error("new_agent is None — aborting handoff")
        return

    await session.update_agent(new_agent)
    await session.generate_reply(instructions="Switched to Transcriber; LLM and TTS disabled.")

Please paste the exact code inside your

handoff_disable_llm_tts

(or wherever

new_agent

is created). If you also created Transcriber without llm/tts parameters, paste its init too. I’ll point out the exact line causing

new_agent

to be None and show the minimal change to fix it. Sources: Agents and handoffs | LiveKit Documentation | Pipeline nodes and hooks | LiveKit Documentation

<src.agents.transcriber_agent.Transcriber object at 0x7a431648ee10> <livekit.agents.voice.agent_session.AgentSession object at 0x7a4305bf2480> I printed the new_agent and session

Thanks — you’ve shown that

new_agent

and

session

are valid objects, so we can focus on the exact call site that triggers the error. Why the error still occurs • The traceback

TypeError: object NoneType can't be used in 'await' expression

means Python tried to

await

something that was None. • Even though your prints show

new_agent

and

session

exist, the awaited expression itself was None. The two most common causes: a. You called an async function that returned None and then tried to

await

that return value. b. You accidentally wrote code that sets the awaited expression to None (for example,

result = session.update_agent(new_agent)

in a sync context and then later doing

await result

after

result

became None). Quick checks to run now 1. Confirm the exact line you await: ◦ It must be:

await session.update_agent(new_agent)

◦ If you have two lines, avoid: ▪︎

result = session.update_agent(new_agent)

followed by

await result

unless

session.update_agent

returns a coroutine (it does) and you did not accidentally overwrite

result

. 1. Ensure the

await

is inside an async function: ◦

async def handoff_disable_llm_tts(...):

◦ And you schedule it from the sync callback with

asyncio.create_task(handoff_disable_llm_tts(session))

. 1. Add a defensive log just before the await:

<http://logger.info|logger.info>("about to await update_agent", extra={"new_agent": new_agent, "session": session})
   if new_agent is None:
       logger.error("new_agent is None - aborting")
       return
   await session.update_agent(new_agent)

This confirms exactly what you are awaiting at runtime. Potential subtle bug patterns • Creating

new_agent

by calling a function that sometimes returns None: ◦ e.g.,

new_agent = maybe_create_agent()

where

maybe_create_agent()

returns None in some branch. • Reassigning the awaited variable earlier in the function (accidental overwrite). • Trying to await the return value of a function that itself performs the await internally and returns None. If you’re happy to paste the exact

handoff_disable_llm_tts

function body (and how you call it from the participant handler), I’ll point to the exact line causing the None and give the minimal change. Based on what you’ve already shown, the fix will likely be one of: • Ensure you

await session.update_agent(new_agent)

directly inside an async function, or • Fix the factory so

new_agent

is never None before the await. Sources: Agents and handoffs | LiveKit Documentation | Transcriber | LiveKit Documentation

@ctx.room.on("participant_connected") def on_participant_connected(participant: rtc.RemoteParticipant) -> None: print(participant) logger.info("participant connected: %s %s", participant.sid, participant.identity) if participant.identity == "identity": print('='*100) print(participant) print('='*100) try: transcriber_agent = Transcriber() #asyncio.create_task(session.update_agent(agent=transcriber_agent)) asyncio.create_task(handoff_disable_llm_tts(session)) except Exception as e: logger.error(f"Failed to update agent: {e}") async def handoff_disable_llm_tts(session: AgentSession): # Instantiate the new agent with LLM and TTS disabled. # Pass the session's STT if you still want transcription. new_agent = Transcriber() print(new_agent) print(session) print("*"*100) # Graceful handoff: update the session's active agent await session.update_agent(new_agent) <src.agents.transcriber_agent.Transcriber object at 0x7a431648ee10> <livekit.agents.voice.agent_session.AgentSession object at 0x7a4305bf2480> **************************************************************************************************** 105510.142 INFO claim_ve…ion_agent participant connected: PA_3k4cEmAKbfup identity {"room": "voice_assistant_room_1848", "pid": 159680, "job_id": "AJ_45yKqMk2UXQS", "room_id": "RM_FeGnfdTuF7Fh"} 105510.144 ERROR asyncio Task exception was never retrieved future: <Task finished name='Task-167' coro=<entrypoint.<locals>.handoff_disable_llm_tts() done, defined at /home/azureuser/AI-development/entrypoint.py:136> exception=TypeError("object NoneType can't be used in 'await' expression")> Traceback (most recent call last): File "/home/azureuser/AI-development/entrypoint.py", line 145, in handoff_disable_llm_tts await session.update_agent(new_agent) TypeError: object NoneType can't be used in 'await' expression {"room": "voice_assistant_room_1848", "pid": 159680, "job_id": "AJ_45yKqMk2UXQS", "room_id": "RM_FeGnfdTuF7Fh"} Traceback (most recent call last): File "/home/azureuser/AI-development/entrypoint.py", line 145, in handoff_disable_llm_tts await session.update_agent(new_agent) TypeError: object NoneType can't be used in 'await' expression

Thanks — you’ve provided exactly the right context. Your prints show

new_agent

and

session

are real objects, so the problem is not that those variables are None. The TypeError happens because the expression you attempted to await evaluated to None. Two very likely causes in this code path 1) update_agent is synchronous in your runtime (returns None) and you awaited it • If

session.update_agent(...)

is a synchronous method that performs the handoff immediately and returns None, doing

await session.update_agent(...)

will trigger this TypeError because you attempted to await a None value. • Some APIs expose both sync and async variants or were changed between versions — confirm whether

update_agent

is async in your installed package. 2) You awaited the return value of another function that returned None (less likely here) • Example: calling a helper that internally awaited

update_agent

and returned None, then `await`ing that helper's return value outside would cause the same error. From your snippet you await

session.update_agent(...)

directly, so this is less likely. How to verify quickly • Print whether update_agent is a coroutine function or not:

import inspect
  <http://logger.info|logger.info>("update_agent iscoroutinefunction=%s iscoroutine=%s",
              inspect.iscoroutinefunction(session.update_agent),
              inspect.iscoroutine(session.update_agent))

• If

inspect.iscoroutinefunction(session.update_agent)

is True, it should be awaited. • If False, it is synchronous and should not be awaited. Correct patterns depending on what you find A) If update_agent is async (must be awaited) • Keep your current structure but ensure the await is inside an async function (which you already have):

async def handoff_disable_llm_tts(session: AgentSession):
      new_agent = Transcriber(llm=None, tts=None, stt=session.stt)
      if new_agent is None:
          logger.error("new_agent is None - aborting")
          return
      <http://logger.info|logger.info>("awaiting async update_agent")
      await session.update_agent(new_agent)
      await session.generate_reply(instructions="Switched to Transcriber; LLM and TTS disabled.")

• Schedule it from the sync room handler:

@ctx.room.on("participant_connected")
  def on_participant_connected(participant):
      if participant.identity == "identity":
          asyncio.create_task(handoff_disable_llm_tts(session))

B) If update_agent is synchronous (do NOT await) • Call it directly inside a task or thread-safe context. Because your room handler is sync, call it in a background task but without awaiting it:

async def handoff_disable_llm_tts(session: AgentSession):
      new_agent = Transcriber(llm=None, tts=None, stt=session.stt)
      if new_agent is None:
          logger.error("new_agent is None - aborting")
          return
      <http://logger.info|logger.info>("calling sync update_agent")
      # CALL IT DIRECTLY (no await) if it's sync
      session.update_agent(new_agent)
      # If generate_reply is async, await it; otherwise call directly
      if inspect.iscoroutinefunction(session.generate_reply):
          await session.generate_reply(instructions="Switched to Transcriber; LLM and TTS disabled.")
      else:
          session.generate_reply(instructions="Switched to Transcriber; LLM and TTS disabled.")

Practical next steps I recommend you run now 1. Add the inspect debug to see whether

session.update_agent

is coroutine or not. 2. If it is synchronous, remove

await

and call it directly from your async task (or call it from the sync handler if safe). 3. If it is asynchronous, keep

await

but ensure the function you

create_task

is the async handoff that awaits it (which you already do). One more defensive improvement • Log right before the await/call and after it (so you see which line throws):

<http://logger.info|logger.info>("new_agent=%s session=%s", new_agent, session)
  <http://logger.info|logger.info>("about to call/await update_agent")
  # await or call based on inspect result
  <http://logger.info|logger.info>("update_agent completed")

If you paste the inspect output (the boolean results) I’ll tell you exactly which variant to use and provide the minimal corrected snippet to drop straight into your code.