sharing complete file here. please let me know if you need more context from __future__ import annotations import asyncio import json import os from livekit import agents, rtc from livekit.agents import (AgentSession, JobContext, MetricsCollectedEvent, RoomInputOptions, metrics, BackgroundAudioPlayer, AudioConfig, BuiltinAudioClip, ConversationItemAddedEvent) from livekit.plugins import noise_cancellation, assemblyai from schemas.constants_and_enums import CallStatus, CallEndedReason from config import ConfigManager from telephone_assistant import CallingAssistant from plugins import PluginFactory from schemas.defaults import get_default_assistant_config, BackgroundSound from services.call_finalization_service import CallFinalizationService from services.egress_service import EgressService from services.session_log_service import SessionLogService, LogComponent, LogLevel from timeout_manager import TimeoutManager from common_settings import settings from livekit.agents.voice.events import ErrorEvent from api_client.call_service_client import CallService from datetime import datetime, timezone from livekit.plugins import groq from livekit import api async def entrypoint(ctx: JobContext): """LiveKit worker entrypoint – wires everything together.""" # Initialize session logging early call_id = None session_logger = None try: await ctx.connect() global_cfg = get_default_assistant_config() agent_cfg = json.loads(os.getenv("AGENT_CONFIG", "{}")) call_cfg = json.loads(ctx.job.metadata or "{}") cfg = ConfigManager(global_cfg).build(agent_cfg, call_cfg) call_id = call_cfg.get("call_id") assistant_id = call_cfg.get("assistant_id") # Initialize session logging with call_id session_logger = SessionLogService( session_id=call_id, max_buffer_size=10000, min_log_level=LogLevel.DEBUG ) # Set session metadata session_logger.set_session_metadata({ "room_name": ctx.room.name, "default_config": global_cfg, "config_from_call_request": call_cfg, "final_config_after_merge" : cfg }) if not call_id: session_logger.log_error(LogComponent.AGENT, "call_id not found in room metadata. Aborting job.") ctx.shutdown() return call_srv = CallService(call_id=call_id) call_srv.update_call(status=CallStatus.DIALING) session_logger.log_info( LogComponent.AGENT, "Session started successfully", room_name=ctx.room.name, call_id=call_id ) # Instantiate services egress_srv = EgressService(session_logger=session_logger, cfg=cfg) session_logger.log_info(LogComponent.AGENT, "Core services initialized") # Build plugins credentials = cfg.get("credentials", []) session_logger.log_info(LogComponent.AGENT, "Building plugins", plugin_count=5) try: plugins = { "stt": PluginFactory.transcriber(cfg["transcriber"], credentials), "llm": PluginFactory.model(cfg["model"], credentials), "tts": PluginFactory.voice(cfg["voice"], credentials), "vad": PluginFactory.vad(), "turn": PluginFactory.turn_detector(), } session_logger.log_info( LogComponent.AGENT, "All plugins created successfully", stt_provider=cfg["transcriber"].get("provider"), llm_provider=cfg["model"].get("provider"), tts_provider=cfg["voice"].get("provider") ) except Exception as e: session_logger.log_error( LogComponent.AGENT, "Failed to create plugins", error=str(e), config=cfg ) raise # Agent session session_logger.log_info(LogComponent.AGENT, "Creating agent session") agent_session = AgentSession( # stt=plugins["stt"], stt=assemblyai.STT(model="universal-streaming-english", api_key=settings.ASSEMBLY_AI_API_KEY), # llm=plugins["llm"], tts=plugins["tts"], llm=groq.LLM(api_key=settings.GROQ_API_KEY, model="llama-3.3-70b-versatile", temperature=0.5, max_completion_tokens=300), vad=plugins["vad"], turn_detection=plugins["turn"], min_interruption_duration=float(cfg.get("stop_speaking_plan").get("voice_seconds")), min_interruption_words=int(cfg.get("stop_speaking_plan").get("num_words")), min_endpointing_delay=0.7, max_endpointing_delay=2, ) session_logger.log_info( LogComponent.AGENT, "Agent session configured", min_interruption_duration=float(cfg.get("stop_speaking_plan").get("voice_seconds")), min_interruption_words=int(cfg.get("stop_speaking_plan").get("num_words")) ) previous_call_status = CallStatus.QUEUED.value async def get_sip_call_status(participant: rtc.RemoteParticipant): nonlocal previous_call_status while True: current_status = participant.attributes.get("sip.callStatus") if current_status != previous_call_status and current_status!= None and previous_call_status != None: session_logger.log_info(LogComponent.AGENT, f"Call status changed from {previous_call_status} to {current_status}") previous_call_status = current_status if current_status == 'active': call_srv.update_call(status=CallStatus.IN_PROGRESS, started_at=datetime.now(timezone.utc)) try: await egress_srv.start_recording(ctx.room.name) except Exception as e: session_logger.log_error(LogComponent.AGENT, "Failed to start recording", error=str(e)) break else: call_srv.update_call(status=current_status) await asyncio.sleep(0.1) # hook participant join @ctx.room.on("participant_connected") def _on_participant_connected(p: rtc.RemoteParticipant): session_logger.log_info( LogComponent.WEBRTC, "Participant connected", participant_id=p.sid, participant_identity=p.identity ) asyncio.create_task(get_sip_call_status(p)) @ctx.room.on("participant_disconnected") def _on_participant_disconnected(p: rtc.RemoteParticipant): session_logger.log_info( LogComponent.WEBRTC, "Participant disconnected", participant_id=p.sid, participant_identity=p.identity ) current_status = p.attributes.get("sip.callStatus") session_logger.log_info( LogComponent.WEBRTC, f"Call status {current_status}", ) dis_connected_reason = p.disconnect_reason session_logger.log_info( LogComponent.WEBRTC, f"Disconnect reason {dis_connected_reason}", ) if current_status in ['ringing', 'dialing'] and dis_connected_reason == 11: call_ended_reason = CallEndedReason.BUSY elif current_status in ['ringing', 'dialing'] and dis_connected_reason == 12: call_ended_reason = CallEndedReason.REJECTED elif current_status == 'hangup': call_ended_reason = CallEndedReason.CALL_RECIPIENT_HUNGUP else: call_ended_reason = CallEndedReason.UNKNOWN finalization_service._ended_reason = call_ended_reason ctx.shutdown() @agent_session.on("error") def on_error(ev: ErrorEvent): if ev.error.recoverable and ev.error.type: return error_map = { "llm_error": (CallEndedReason.LLM_ERROR, "LLM error occurred"), "stt_error": (CallEndedReason.STT_ERROR, "STT error occurred"), "tts_error": (CallEndedReason.TTS_ERROR, "TTS error occurred"), } reason, message = error_map.get(ev.error.type, (CallEndedReason.UNKNOWN, "Unrecoverable error occurred")) log_level = LogLevel.CRITICAL if reason == CallEndedReason.UNKNOWN else LogLevel.ERROR session_logger.log( LogComponent.AGENT, log_level, message, error_message=str(ev.error.error), error_type=ev.error.type ) finalization_service._ended_reason = reason ctx.shutdown() # Setup metrics collection usage_collector = metrics.UsageCollector() @agent_session.on("metrics_collected") def _on_metrics_collected(ev: MetricsCollectedEvent): usage_collector.collect(ev.metrics) metric_type = ev.metrics.type if metric_type not in ["vad_metrics", "stt_metrics"]: session_logger.log_info( LogComponent.AGENT, f"{metric_type} metrics collected", metrics_data=ev.metrics.model_dump() ) @agent_session.on("conversation_item_added") def on_conversation_item_added(event: ConversationItemAddedEvent): role = event.item.role if role == "user": session_logger.log_info( LogComponent.STT, "User speech transcribed.", text=event.item.text_content ) elif role == "assistant": session_logger.log_info( LogComponent.LLM, "LLM genearted Response", text=event.item.text_content ) end_phrases = [ "bye", "goodbye", "we look forward to your participation", "have a great day ahead", "best of luck", "have a great day" ] if any(phrase in event.item.text_content.lower() for phrase in end_phrases): finalization_service._ended_reason = CallEndedReason.ASSISTANT_SAID_END_CALL_PHRASE ctx.shutdown() # Instantiate the finalization service and register the shutdown callback finalization_service = CallFinalizationService( call_id=call_id, call_service=call_srv, agent_session=agent_session, egress_service=egress_srv, usage_collector=usage_collector, job_context=ctx, cfg=cfg, session_logger=session_logger, ) session_logger.log_info(LogComponent.FINALIZATION, "Finalization service initialized") # Timeout manager session_logger.log_info(LogComponent.TIMEOUT, "Initializing timeout manager") tm = TimeoutManager( idle_timeout_s=cfg.get("message_plan").get("idle_timeout_seconds"), idle_timeout_on_wait_s=cfg.get("message_plan").get("idle_timeout_on_wait_seconds"), max_s=cfg.get("max_duration_seconds"), idle_messages=cfg.get("message_plan").get("idle_messages"), idle_maximum_messages=cfg.get("message_plan").get("idle_message_max_spoken_count"), idle_timeout_message=cfg.get("message_plan").get("idle_timeout_message"), max_timeout_message=cfg.get("message_plan").get("max_timeout_message"), agent_session=agent_session, session_logger=session_logger, finalization_service=finalization_service ) session_logger.log_info( LogComponent.TIMEOUT, "Timeout manager configured", idle_timeout=cfg.get("message_plan").get("idle_timeout_seconds"), max_duration=cfg.get("max_duration_seconds") ) asyncio.create_task(tm.watch()) # Interview Assistant prompt_variable_values = cfg.get("assistant_overrides", {}).get("variable_values", {}) phone_number = prompt_variable_values.pop("phone_number") instructions = cfg.get("model").get("messages")[0].get("content") session_logger.log_info( LogComponent.AGENT, "Creating interview assistant", has_instructions=bool(instructions) ) assistant = CallingAssistant( instructions=instructions, prompt_variable_values=prompt_variable_values, finalization_service=finalization_service, timeout_mgr=tm, session_logger=session_logger, call_service=call_srv ) session_logger.log_info(LogComponent.AGENT, "Starting agent session") try: if not phone_number: session_logger.log_error( LogComponent.AGENT, "Phone number not found in prompt variable values", prompt_variable_values=prompt_variable_values ) ctx.shutdown() if assistant_id == settings.LIVEKIT_ASSISTANT_ID_RESUME_SHORTLISTED: sip_trunk_id = settings.SIP_OUTBOUND_TRUNK_ID_RESUME_SHORTLISTED elif assistant_id == settings.LIVEKIT_ASSISTANT_ID_SCHEDULED_MISSED: sip_trunk_id = settings.SIP_OUTBOUND_TRUNK_ID_SCHEDULED_MISSED elif assistant_id == settings.LIVEKIT_ASSISTANT_ID_OUTBOUND_RESUME_SHORTLISTED: sip_trunk_id = settings.SIP_OUTBOUND_TRUNK_ID_OUTBOUND_RESUME_SHORTLISTED else: sip_trunk_id = settings.SIP_OUTBOUND_TRUNK_ID_DEFAULT lkapi = api.LiveKitAPI() sip_participant = await lkapi.sip.create_sip_participant( api.CreateSIPParticipantRequest( room_name=ctx.room.name, sip_trunk_id=sip_trunk_id, sip_call_to=f"+91{phone_number}", participant_identity=f"sip_callee_{phone_number}", wait_until_answered=True ) ) session_logger.log_info( LogComponent.AGENT, f"SIP participant created {sip_participant}", ) # try: # await egress_srv.start_recording(ctx.room.name) # except Exception as e: # session_logger.log_error(LogComponent.AGENT, "Failed to start recording", error=str(e)) except Exception as e: session_logger.log_error(LogComponent.AGENT, "Failed to create SIP participant", error=str(e)) finally: await lkapi.aclose() # Start session await agent_session.start( room=ctx.room, agent=assistant, room_input_options=RoomInputOptions( audio_enabled=True, text_enabled=False, noise_cancellation=noise_cancellation.BVC(), ), ) session_logger.log_info(LogComponent.AGENT, "Agent session started successfully") except Exception as e: error_msg = f"Fatal error in entrypoint: {str(e)}" if session_logger: session_logger.log_critical( LogComponent.AGENT, error_msg, error_type=type(e).__name__, call_id=call_id ) # Set the error reason before shutting down if 'finalization_service' in locals(): finalization_service._ended_reason = CallEndedReason.SERVER_ERROR # Ensure cleanup even on error if ctx: if 'finalization_service' in locals() and len(ctx._shutdown_callbacks) == 0: ctx.add_shutdown_callback(finalization_service.finalize_call_record) ctx.shutdown() raise ctx.add_shutdown_callback(finalization_service.finalize_call_record) ###################################################################### # CLI helper # ###################################################################### def main(): agents.cli.run_app( agents.WorkerOptions( entrypoint_fnc=entrypoint, agent_name=settings.LIVEKIT_AGENT_WORKER_NAME, port=8082 ) ) if __name__ == "__main__": main()