blue-account-81612
07/08/2025, 5:25 PMtall-airline-16756
07/08/2025, 5:25 PMtall-airline-16756
07/08/2025, 5:25 PMimport asyncio
import logging
from livekit.agents import Agent, AgentSession, CloseEvent, JobContext, WorkerOptions, cli
logger = logging.getLogger("my-worker")
logger.setLevel(<http://logging.INFO|logging.INFO>)
class MyAgent(Agent):
def __init__(self):
super().__init__(instructions="You are a helpful assistant.")
async def on_session_close(event: CloseEvent):
session_id = event.session.id
<http://logger.info|logger.info>(f"Session {session_id} closed, reason: {event.reason}")
# Trigger webhook here
async def trigger_webhook(session_id):
try:
webhook_url = "<https://your-webhook-url.com/hangup>"
data = {"session_id": session_id, "event": "hangup"}
headers = {"Content-Type": "application/json"}
await <http://httpx.post|httpx.post>(webhook_url, json=data, headers=headers)
except Exception as e:
logger.error(f"Failed to trigger webhook: {e}")
await trigger_webhook(session_id)
async def entrypoint(ctx: JobContext):
session = AgentSession()
@session.on("close")
def handle_close(event: CloseEvent):
asyncio.create_task(on_session_close(event))
await session.start(agent=MyAgent(), room=ctx.room)
await ctx.connect()
if __name__ == "__main__":
cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))
This setup ensures that when the call ends, the agent worker is removed automatically, and your webhook gets notified with the call session ID and hangup event.
Remember to handle authentication and failure cases in your webhook implementation.
Sources: .github/next-release/changeset-12012403.md