https://livekit.io logo
Join Slack
Powered by
``` await self._room.local_participant.publish_...
# ask-ai
n
Copy code
await self._room.local_participant.publish_transcription(transcription)

  File "/usr/local/lib/python3.11/site-packages/livekit/rtc/participant.py", line 285, in publish_transcription

    raise PublishTranscriptionError(cb.publish_transcription.error)

livekit.rtc.participant.PublishTranscriptionError: failed to send transcription, room closed: channel closed

WARNING:livekit.agents:failed to publish transcription

Traceback (most recent call last):

  File "/usr/local/lib/python3.11/site-packages/livekit/agents/voice/room_io/_output.py", line 429, in capture_text

    self._writer = await self._create_text_writer()

                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/lib/python3.11/site-packages/livekit/agents/voice/room_io/_output.py", line 405, in _create_text_writer

    return await self._room.local_participant.stream_text(

           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/lib/python3.11/site-packages/livekit/rtc/participant.py", line 578, in stream_text

    await writer._send_header()

  File "/usr/local/lib/python3.11/site-packages/livekit/rtc/data_stream.py", line 189, in _send_header

    raise ConnectionError(cb.send_stream_header.error)

ConnectionError: engine: connection error: engine is closed

WARNING:livekit.agents:failed to publish transcription

Traceback (most recent call last):

  File "/usr/local/lib/python3.11/site-packages/livekit/agents/voice/room_io/_output.py", line 300, in _publish_transcription

    await self._room.local_participant.publish_transcription(transcription)

  File "/usr/local/lib/python3.11/site-packages/livekit/rtc/participant.py", line 285, in publish_transcription

    raise PublishTranscriptionError(cb.publish_transcription.error)

livekit.rtc.participant.PublishTranscriptionError: failed to send transcription, room closed: channel closed

WARNING:livekit.agents:failed to publish transcription

Traceback (most recent call last):

  File "/usr/local/lib/python3.11/site-packages/livekit/agents/voice/room_io/_output.py", line 429, in capture_text

    self._writer = await self._create_text_writer()

                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/lib/python3.11/site-packages/livekit/agents/voice/room_io/_output.py", line 405, in _create_text_writer

    return await self._room.local_participant.stream_text(

           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/lib/python3.11/site-packages/livekit/rtc/participant.py", line 578, in stream_text

    await writer._send_header()

  File "/usr/local/lib/python3.11/site-packages/livekit/rtc/data_stream.py", line 189, in _send_header

    raise ConnectionError(cb.send_stream_header.error)

ConnectionError: engine: connection error: engine is closed

WARNING:livekit.agents:failed to publish transcription

Traceback (most recent call last):

  File "/usr/local/lib/python3.11/site-packages/livekit/agents/voice/room_io/_output.py", line 300, in _publish_transcription

    await self._room.local_participant.publish_transcription(transcription)

  File "/usr/local/lib/python3.11/site-packages/livekit/rtc/participant.py", line 285, in publish_transcription

    raise PublishTranscriptionError(cb.publish_transcription.error)

livekit.rtc.participant.PublishTranscriptionError: failed to send transcription, room closed: channel closed

what can i do ?
Copy code
await self.room.disconnect()
api_client = api.LiveKitAPI()
try:
    await api_client.room.delete_room(
        DeleteRoomRequest(room=self.room.name)
    )
    <http://logging.info|logging.info>("☎️ Room deleted – call ended")
except Exception as e:
    logging.error(f"☎️ Failed to delete room: {e}")
finally:
    await api_client.aclose()
t
Thanks for reaching out, Darshak. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — I reviewed the error traces you shared. You’re getting repeated: • 
PublishTranscriptionError: failed to send transcription, room closed: channel closed
• 
ConnectionError: engine: connection error: engine is closed
These errors mean you’re attempting to send transcription/text data after the room/engine has already been closed. Below are focused actions and code patterns to stop those errors and cleanly shut down. What I’m answering: how to stop publishing transcriptions after the room is closed and how to safely disconnect/delete the room. 1) Stop publishing once the room/engine is closed • Add an explicit guard before any call that publishes text:
Copy code
if not self._room or not getattr(self._room, "is_connected", False):
    # skip publishing — room is closed or already disconnecting
    return
await self._room.local_participant.publish_transcription(transcription)
• Use this wrapper everywhere you would call 
publish_transcription()
or 
stream_text()
. 2) Cancel background tasks that publish transcriptions before disconnect • Long-running STT, capture, or agent tasks can try to publish after you call 
disconnect()
. Cancel those tasks first, wait for them to finish, then disconnect:
Copy code
# cancel tasks
if self._stt_task:
    self._stt_task.cancel()
if self._capture_text_task:
    self._capture_text_task.cancel()

# optionally await them with timeout to let cleanup run
try:
    await asyncio.wait_for(self._stt_task, timeout=1.0)
except Exception:
    pass

# then disconnect the room
if self._room and getattr(self._room, "is_connected", False):
    await self._room.disconnect()
    self._room = None  # prevent accidental reuse
3) React to room lifecycle events and clean up immediately • Listen for a room 
disconnected
(or similar) event and cancel/pause publishing there so unexpected disconnects don’t trigger further publishes:
Copy code
@self._room.on("disconnected")
def on_disconnected():
    # cancel or flag tasks to stop publishing
    if self._capture_text_task:
        self._capture_text_task.cancel()
4) Order of operations for shutdown (safe sequence) 1. Cancel/pause any STT/agent tasks that generate transcriptions. 2. Ensure no further publish calls will be made (guards or set a flag). 3. Call 
await self.room.disconnect()
. 4. Use the LiveKit API to delete the room:
Copy code
api_client = api.LiveKitAPI()
try:
    await api_client.room.delete_room(DeleteRoomRequest(room=self.room.name))
finally:
    await api_client.aclose()
5) Optional: use 
delete_room_on_close
for agent sessions • If you create agent sessions, 
delete_room_on_close=True
can auto-delete the room when the session ends (useful if agent is the last participant). This avoids manually calling 
delete_room()
in many flows. 6) Logging to diagnose race conditions • Add clear logs when you cancel tasks, call disconnect, and when publish attempts occur. This helps identify if publishes are coming from tasks you forgot to cancel. Why these fixes work • The errors occur because once the underlying engine/connection is closed there is no channel to send data. Preventing further publishes (guards, cancelling tasks, listening for disconnect events) stops attempts to write to closed channels and removes the exceptions. If you want, I can: • Show a small complete shutdown example that integrates task cancellation, guards, disconnect, and room deletion. • Help inspect where your transcription publish calls are scheduled so we can add the correct cancellations/guards. Which would you prefer: a minimal shutdown snippet you can drop into your code, or help tracing where background tasks might still be running in your current codebase? Sources: Agent session | LiveKit Documentation
n
so how can i cancel all taskes before doing
Copy code
api_client.room.delete_room(DeleteRoomRequest(room=self.room.name))
finally:
    await api_client.aclose()
?
t
Good question — below is a clear, minimal pattern you can use to cancel all running asyncio tasks that might publish transcriptions, wait for them to clean up, then safely call 
delete_room
. I’ll show (A) how to track tasks, (B) how to cancel & wait for them, and (C) a safe full shutdown sequence including your 
delete_room
call. What I’m answering: how to cancel all tasks before calling 
api_client.room.delete_room(DeleteRoomRequest(room=self.room.name))
. A. Track tasks you start • Keep a set of active tasks so you can cancel them later.
Copy code
self.active_tasks = set()

def start_background_task(coro):
    task = asyncio.create_task(coro)
    self.active_tasks.add(task)
    task.add_done_callback(lambda t: self.active_tasks.discard(t))
    return task
• Use 
start_background_task()
for any STT/capture/publish tasks (so they’re tracked automatically). B. Cancel and await tasks safely • Cancel all tracked tasks, then wait for them to finish cleanup. Awaiting lets those tasks handle 
CancelledError
and do their own cleanup (e.g., stop STT pipelines, close writers).
Copy code
async def cancel_all_tasks(timeout: float = 5.0):
    # Cancel tasks that are still running
    for task in list(self.active_tasks):
        if not task.done():
            task.cancel()

    if not self.active_tasks:
        return

    # Wait for tasks to complete/cancel; ignore exceptions except log them
    done, pending = await asyncio.wait(self.active_tasks, timeout=timeout)
    for t in done:
        try:
            # re-raise exceptions other than CancelledError so you can log them
            t.result()
        except asyncio.CancelledError:
            pass
        except Exception as e:
            logger.exception("Background task ended with error: %s", e)

    # If anything still pending, optionally force them (or log)
    if pending:
        logger.warning("Some tasks did not finish within timeout: %s", pending)
C. Full safe shutdown and room deletion sequence • 1) Stop/cancel tasks that may publish transcription • 2) Disconnect the local room (so LiveKit client stops) • 3) Delete the room via the API • 4) Close API client
Copy code
async def cleanup_and_delete_room(self):
    api_client = api.LiveKitAPI()  # your existing client

    # 1) Cancel app-level background tasks (STT, capture, etc.)
    await cancel_all_tasks(timeout=5.0)

    # 2) Optionally, ensure agent/session shutdown (if using sessions)
    # if self.session:
    #     await self.session.aclose()  # or await self.session.shutdown(...) as appropriate

    # 3) Disconnect local room if still connected
    if self.room and getattr(self.room, "is_connected", False):
        try:
            await self.room.disconnect()
        except Exception as e:
            logger.warning("Error disconnecting room: %s", e)
        finally:
            self.room = None  # avoid accidental reuse

    # 4) Delete the room via API
    try:
        await api_client.room.delete_room(DeleteRoomRequest(room=self.room_name))
        <http://logger.info|logger.info>("Room deleted – call ended")
    except Exception as e:
        logger.error("Failed to delete room: %s", e)
    finally:
        await api_client.aclose()
D. Extra guards to prevent races • Add a boolean shutdown flag so any in-flight publish checks it before sending:
Copy code
self.shutting_down = False

# when shutting down:
self.shutting_down = True

# wrapper used when publishing:
if self.shutting_down or not self.room or not getattr(self.room, "is_connected", False):
    return
await self.room.local_participant.publish_transcription(transcription)
• This avoids tasks starting new publish attempts while you cancel them. E. Logging and short timeouts • Give tasks a short timeout (e.g., 5 seconds) to clean up; log any that remain. That helps identify stray tasks in subsequent runs. If you’d like, I can: • Convert this into a drop-in class method for your agent. • Show how to modify a typical transcription loop to respect 
self.shutting_down
and handle 
CancelledError
. Which would help you most — a direct patch for your agent’s transcription loop, or a small self-contained shutdown helper you can import? Sources: SIP Lifecycle Management Agent | LiveKit Documentation | Text and transcriptions | LiveKit Documentation | Agents telephony integration | LiveKit Documentation | Job lifecycle | LiveKit Documentation | Agent speech and audio | LiveKit Documentation
n
want to give tool to llm such that when agent want to cut call it can do it, making outbound call via twilio
t
Great — you want the LLM agent to be able to hang up an outbound Twilio call. The simplest reliable way is to expose a tool the agent can call that deletes the LiveKit room (LiveKit will send the SIP BYE to Twilio and the call will end). Below is a compact, safe pattern you can implement. What I’m answering: how to give the LLM a callable tool that ends the Twilio outbound call and how to do it safely (graceful finish + cancel tasks). 1) Tool to end the call (core) • Implement a function tool the LLM can call that: ◦ optionally plays a final message (or waits for playout), ◦ deletes the LiveKit room using the Server API, ◦ returns a success/failure message to the agent. Example (Python, for use with livekit-agents):
Copy code
from livekit import api
from livekit.agents import get_job_context, function_tool

@function_tool()
async def hangup_call(self, ctx):
    """Tool for the agent to end the current call by deleting the room."""
    job_ctx = get_job_context()
    if job_ctx is None:
        return "No active job context found."

    # Optional: wait for session playout so final speech finishes
    try:
        await ctx.session.wait_for_playout()
    except Exception:
        pass

    try:
        await job_ctx.api.room.delete_room(
            api.DeleteRoomRequest(room=job_ctx.room.name)
        )
        return "Call ended successfully."
    except Exception as e:
        return f"Failed to end call: {e}"
2) Make the tool available to the agent • Register 
hangup_call
in the agent’s tools list and document when to use it (e.g., user says "goodbye" or agent decides to end). • Example instruction: "Use hangup_call() to terminate the call when the user or agent decides to finish." 3) Graceful behavior & safety • Before hangup, have the agent confirm or speak a final message. Use 
ctx.session.wait_for_playout()
so the message is delivered before deletion. • Use a short confirmation step in the agent flow if accidental hangups are a concern. 4) Cancel background tasks and avoid race conditions • Ensure any background tasks that may publish transcription or audio are cancelled before/when deleting the room (prevents "room closed"/"engine closed" errors). • Use a tracked-task pattern: ◦ Add tasks with a helper:
Copy code
self.active_tasks = set()
    def start_task(coro):
        t = asyncio.create_task(coro)
        self.active_tasks.add(t)
        t.add_done_callback(lambda _: self.active_tasks.discard(t))
        return t
• On shutdown/hangup:
Copy code
async def cancel_all_tasks(timeout=5.0):
        for t in list(self.active_tasks):
            if not t.done():
                t.cancel()
        if self.active_tasks:
            await asyncio.wait(self.active_tasks, timeout=timeout)
• Call 
await cancel_all_tasks()
before calling 
delete_room()
inside the tool (or ensure the agent framework has already stopped I/O). 5) Full safe sequence inside your tool (recommended) 1. Set a shutting-down flag (so any in-flight publish checks it). 2. Cancel/await background tasks or ensure session shutdown. 3. Wait for final speech playout (optional). 4. Disconnect local client if needed (optional). 5. Call 
delete_room(...)
. 6. Close API client. 6) Notes about Twilio / SIP • You do not need to call Twilio directly to end the SIP session — deleting the LiveKit room triggers a SIP BYE to Twilio. • Ensure your Twilio trunk settings are appropriate for transfers/hangups if you rely on particular caller-id or transfer behaviors. If you want, I can: • Provide a drop-in 
hangup_call
tool that includes task cancellation and a final playout wait. • Show how to register it with your agent and a sample prompt/instruction so the LLM knows when to call it. Which would you prefer — the complete drop-in tool with task cancellation, or just a minimal tool and guidance for integrating it into your existing agent? Sources: Handling payments with PCI compliance | LiveKit | Agents telephony integration | LiveKit Documentation | Preserve caller ID during SIP transfers with Twilio | LiveKit
2 Views
Open in Slack
PreviousNext
Powered by