https://livekit.io logo
Join Slack
Powered by
How do i manually disconnects the call by calling ...
# ask-ai
a
How do i manually disconnects the call by calling a tool and also calls the add_shutdown_callback ?
✅ 1
t
Thanks for reaching out, Manish. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — I’m responding about how to manually disconnect a call by calling a tool and ensure your 
add_shutdown_callback
handlers run. Short answer • Use the prebuilt 
EndCallTool
to disconnect a call; it triggers agent session shutdown and will execute any shutdown callbacks you registered. • If you need custom logic, call 
ctx.room.disconnect()
or 
session.shutdown()
/ 
await session.aclose()
directly; those also invoke registered shutdown callbacks. • Register cleanup via 
ctx.add_shutdown_callback()
in your entrypoint so your cleanup runs when the session shuts down. What to do (two common approaches) 1) Use the EndCallTool (recommended for agent-driven/agent-tool flows) • Add the 
EndCallTool
to your agent’s tool list. • When the tool is invoked, it will generate the final response, shut down the session, and run all shutdown callbacks you registered with 
ctx.add_shutdown_callback()
. • This is the simplest option when you want a tool the agent can call to end the call and run cleanup. 2) Manually disconnect from code (for programmatic or custom triggers) 1. Register your cleanup callback in the entrypoint:
Copy code
async def on_shutdown():
        logger.info("Post-conversation cleanup: webhooks, logs, etc.")
        # put your cleanup logic here

    ctx.add_shutdown_callback(on_shutdown)
1. When you want to end the call: ◦ Either call 
ctx.room.disconnect()
(disconnects the room) ◦ Or call 
session.shutdown(drain=True)
to let in-flight speech complete, or 
await session.aclose()
for an awaitable shutdown 1. Shutdown will trigger the callbacks you registered with 
ctx.add_shutdown_callback()
. Notes and useful details • 
session.shutdown(drain=True)
ensures any ongoing speech completes before termination. • 
await session.aclose()
can be used when you need an awaitable shutdown. • The 
EndCallTool
automates final response generation, room deletion (optional), and session shutdown so you don’t need to call 
session.shutdown()
yourself when you use it. If you’d like, tell me which environment you’re using (agent with tools or custom server code) and paste the minimal code you currently have — I can give code examples tailored to your setup. Sources: Prebuilt tools | LiveKit Documentation | EndCallTool | LiveKit Documentation | Collect recording consent with tasks | LiveKit Documentation | Job lifecycle | LiveKit Documentation | How to detect and handle silence during calls, with interactive elements
Thanks — I’m looking at the 
close_the_session_delete_room
function you posted and will comment on whether it follows the telephony cleanup pattern and how to make sure your 
add_shutdown_callback
handlers run reliably. High-level assessment • Your sequence is correct for a graceful shutdown: 1) initiate session shutdown (so RecorderIO stops and session-internal resources wind down), 2) wait briefly for shutdown to take effect, then 3) delete the LiveKit room so the SIP caller is disconnected and any shutdown callbacks run. • The idempotency guard is good — it prevents duplicate termination attempts. Recommendations and clarifications 1) Use of 
session.shutdown(drain=True)
• Calling 
self.session.shutdown(drain=True)
is appropriate when you want to allow in-flight speech to complete. • Note that 
shutdown()
in your environment is non-blocking, so your subsequent 
await asyncio.sleep(1.0)
is sensible to give RecorderIO time to stop. If you can observe reliably when RecorderIO is stopped (e.g., by polling a flag or awaiting an async-complete API), prefer that instead of a fixed sleep to avoid races or unnecessary delay. 2) Handling RuntimeError from shutdown • Your RuntimeError handling is good to treat already-shutdown sessions as non-fatal. • Keep the exact string checks you have (e.g., "no activity context found" or "isn't running") if those are the exact messages you observe; otherwise consider a conservative fallback to log and continue rather than re-raising for unknown shutdown errors. 3) Ensuring 
ctx.add_shutdown_callback
runs • Deleting the room (
job_ctx.api.room.delete_room(...)
) triggers the agent/session shutdown flow, which runs the callbacks you registered via 
ctx.add_shutdown_callback()
. • Make sure the callbacks you registered are asynchronous functions (or return awaitables) and that they themselves handle errors. If a shutdown callback raises uncaught exceptions, it may interrupt remaining cleanup tasks — wrap callback bodies in try/except and log failures. 4) Idempotency and error resilience around room deletion • Room deletion can fail transiently (network, API). Consider retrying the delete call with a short exponential backoff if it returns transient errors. • If delete_room returns a not-found response, treat it as success (room already deleted). 5) Await vs non-await behavior • You call 
self.session.shutdown(drain=True)
without awaiting; that matches your note that shutdown is non-blocking. If your runtime exposes an awaitable like 
await session.aclose()
or 
await session.shutdown()
then using that will provide a deterministic await for completion. If not available, prefer an observable completion condition rather than a fixed sleep. 6) Make callbacks robust and observable • Inside your shutdown callbacks do: ◦ Wrap logic in try/except and log errors. ◦ Make long-running uploads or webhook calls cancellable or bounded by timeouts so shutdown doesn’t hang indefinitely. ◦ Optionally record shutdown progress/status so you can audit whether uploads completed. 7) Example small refinements (pseudocode) 1. Check or await RecorderIO stopped if available instead of fixed sleep. 2. Retry room.delete on transient errors. 3. Timebox shutdown callbacks (via timeout) if they perform network I/O. If you’d like, I can: • Suggest concrete retry/backoff code for 
delete_room
. • Show an example of wrapping a shutdown callback with a timeout and safe error handling. • Turn your function into a version that awaits a deterministic shutdown if 
session.aclose()
(or an awaitable shutdown) exists in your runtime. Which of those would be most helpful right now — retry logic for delete, an example safe shutdown-callback wrapper, or converting to an awaitable shutdown pattern? , with interactive elements
a
converting to an awaitable shutdown pattern
t
Thanks — I’m responding about converting your non-awaitable shutdown sequence into an awaitable shutdown pattern so cleanup is deterministic and your 
ctx.add_shutdown_callback
handlers run reliably. Below I give a concise explanation, a recommended pattern, and a concrete example you can adapt. Why use an awaitable shutdown • Awaiting shutdown gives you a deterministic point when session-internal cleanup (RecorderIO stopping, agent shutdown callbacks) has finished. • It avoids race conditions and arbitrary sleeps, and makes it easier to sequence room deletion, retries, and post-shutdown work. Recommended pattern (summary) 1. Register shutdown callbacks with 
ctx.add_shutdown_callback()
as async functions that handle their own errors and timeouts. 2. If your runtime exposes an awaitable shutdown (e.g., 
await session.aclose()
or 
await session.shutdown()
), call and await that to guarantee the session is shutdown. 3. After the awaitable shutdown returns, retry-safe delete the LiveKit room (or treat 404 as success). 4. Use timeouts around long-running callbacks or the awaitable shutdown if you must bound total shutdown time. Example (adapt to your APIs) • This example assumes: ◦ 
self.session.aclose()
is an awaitable that completes when session shutdown and registered shutdown callbacks have finished. ◦ 
job_ctx.api.room.delete_room(room_name)
is an async call to delete the room. ◦ Use 
asyncio.wait_for
to timebox operations and 
tenacity
-style retry (shown as a simple retry loop).
Copy code
import asyncio
import logging
from typing import Callable

logger = logging.getLogger(__name__)

async def safe_shutdown_and_delete_room(self, job_ctx, room_name: str, shutdown_timeout: float = 10.0):
    """
    Await the session shutdown deterministically, then delete the room with retry/backoff.
    """

    if getattr(self, "_is_shutting_down", False):
        logger.info("Shutdown already in progress; returning early.")
        return

    self._is_shutting_down = True

    # 1) Await the session shutdown (awaitable)
    try:
        # Timebox the awaitable shutdown so we don't hang forever
        await asyncio.wait_for(self.session.aclose(), timeout=shutdown_timeout)
        logger.info("Session awaitable shutdown completed.")
    except asyncio.TimeoutError:
        logger.warning("Timed out waiting for session shutdown; proceeding with room deletion.")
    except RuntimeError as e:
        # Treat already-shutdown as success; log and continue.
        logger.info("Session shutdown RuntimeError; treating as shutdown complete: %s", e)
    except Exception as e:
        logger.exception("Unexpected error while awaiting session shutdown: %s", e)

    # 2) Retry-safe room deletion
    max_retries = 3
    backoff_base = 0.5
    for attempt in range(1, max_retries + 1):
        try:
            await job_ctx.api.room.delete_room(room_name)
            logger.info("Room %s deleted successfully.", room_name)
            break
        except Exception as e:
            # If the API returns not-found, treat as success
            if "not found" in str(e).lower():
                logger.info("Room %s already deleted.", room_name)
                break
            logger.warning("Attempt %d to delete room failed: %s", attempt, e)
            if attempt == max_retries:
                logger.exception("Failed to delete room %s after %d attempts.", room_name, max_retries)
            else:
                await asyncio.sleep(backoff_base * (2 ** (attempt - 1)))

    # 3) Final cleanup/logging (non-blocking)
    try:
        # if you have final metrics or state writes, do them here (bounded by timeout if needed)
        pass
    except Exception:
        logger.exception("Error during final cleanup.")
Making shutdown callbacks robust • Register callbacks as async functions and wrap their body with try/except; use timeouts for network calls:
Copy code
async def on_shutdown():
    try:
        await asyncio.wait_for(send_webhook(...), timeout=5.0)
    except asyncio.TimeoutError:
        logger.warning("Webhook timed out during shutdown")
    except Exception:
        logger.exception("Shutdown callback failed")
ctx.add_shutdown_callback(on_shutdown)
If your runtime does not expose an awaitable shutdown • If 
session.aclose()
or an awaitable 
shutdown()
is not available, emulate an awaitable by: a. Exposing an async-completion event in the session code that is set once internal shutdown and callbacks finish (e.g., 
self._shutdown_complete_event.set()
). b. Await that event in your shutdown helper: 
await self._shutdown_complete_event.wait()
. • Alternatively, modify the lower-level shutdown implementation to return an awaitable that completes after running 
ctx
shutdown callbacks. Next steps I can help with • Modify the snippet to match your exact API names (e.g., 
session.shutdown()
vs 
session.aclose()
). • Add a concrete retry/backoff helper using 
tenacity
. • Show how to expose and await a custom shutdown completion event if your runtime lacks an awaitable shutdown. Which would you like me to generate next — adapt the example to your exact function names, or show the custom event-based awaitable pattern? , with interactive elements
2 Views
Open in Slack
PreviousNext
Powered by