o ip@s@UddlmZddlZddlZddlZddlZddlZddlZddlZ ddl Z ddl m Z m Z ddlmZddlmZmZddlmZddlmZmZddlmZddlZdd lmZmZdd lmZdd l m!Z!m"Z"d d l#m$Z$d dl%m&Z&d dl'm(Z(d dl)m*Z*m+Z+m,Z,d dl-m.Z.d dl/m0Z0m1Z1m2Z2d dl3m4Z4ej5ddZ6erd dl7m8Z8d dl9m:Z:m;Z;d dle>Z?eGdddeZ@GdddeAeZBeGd d!d!ZCeGd"d#d#ZDejEjFejEjGgZHd$eId%<Gd&d'd'ejJZKGd(ddZLd6d-d.ZMGd/d0d0ZNGd1d2d2ZOeGd3d4d4ZPdS)7) annotationsN)Callable Coroutine) dataclass)Enumunique)Path) TYPE_CHECKINGAny)urlparse)apirtc)Claims)agentmodels)logger)Tagger)_upload_session_report)_BufferingHandler_setup_cloud_tracer_shutdown_telemetry) NotGivenOr) http_contextis_givenwait_for_participant)is_cloud JobContextagents_job_context)InferenceExecutor) AgentSessionRecordingOptions SessionReportreturncCstd}|dur td|S)NzHno job context found, are you running this code inside a job entrypoint?)_JobContextVarget RuntimeError)ctxr)T/var/www/html/livekit_bhavya/venv/lib/python3.10/site-packages/livekit/agents/job.pyget_job_context7s r+c@seZdZdZdZdS)JobExecutorTypeprocessthreadN)__name__ __module__ __qualname__PROCESSTHREADr)r)r)r*r,Dsr,c@seZdZdZdZdZdZdS) AutoSubscribe subscribe_allsubscribe_none audio_only video_onlyN)r/r0r1 SUBSCRIBE_ALLSUBSCRIBE_NONE AUDIO_ONLY VIDEO_ONLYr)r)r)r*r4Js r4c@s2eZdZUded<ded<ded<dZded<dS)JobAcceptArgumentsstrnameidentitymetadataNdict[str, str] | None attributes)r/r0r1__annotations__rCr)r)r)r*r=Qs r=c@s>eZdZUded<ded<ded<ded<ded<d ed <d S) RunningJobInfor=accept_arguments agent.Jobjobr>urltoken worker_idboolfake_jobNr/r0r1rDr)r)r)r*rEYs  rEz#list[rtc.ParticipantKind.ValueType]DEFAULT_PARTICIPANT_KINDScs,eZdZdZd fdd Zdd d ZZS)_ContextLogFieldsFilterzGFilter that adds job context fields to log records without overwriting.job_ctxrr$Nonecst||_dSN)super__init__rQ)selfrQ __class__r)r*rUls  z _ContextLogFieldsFilter.__init__recordlogging.LogRecordrLcCsn|jjjtjkrzt}Wn tyYdSw||jkrdS|jjD]\}}t ||s4t |||q%dSNT) rQproc executor_typer,r2r+r' _log_fieldsitemshasattrsetattr)rVrYr(keyvaluer)r)r*filterps     z_ContextLogFieldsFilter.filter)rQrr$rR)rYrZr$rL)r/r0r1__doc__rUrd __classcell__r)r)rWr*rPisrPc@seZdZedejgedfZdddZdddZ dddZ dddZ dddZ dddZ dddZdd d!Zedd#d$Zedd%d&Zedd(d)Zddd.d/Zejdd1d2Zedd3d4Zedd6d7Zedd9d:Zedd;d<Zedd>d?ZeddAdBZeddCdDZeddFdGZej ddIdGZddLdMZ!d*e"dNddTdUZ#d*e$j%d*dVdd]d^Z&dddadbZ'dcddddkdlZ( mdddrdsZ)dddvdwZ*e"dxdd}d~Z+dddZ,dddZ-dddZ.d*S)r)NNNr\ JobProcessinforEroomrtc.Room on_connectCallable[[], None] on_shutdownCallable[[str], None]inference_executorrr$rRc Cs||_||_||_||_||_g|_g|_ttt t j ft j df|_tt j t|_|jd|j||_i|_t||_g|_d|_t|_ddlm}|}|j r^|j!|_"nt#|jj$|_"d|_%t &|_'t(|_)d|_*d|_+dS)Nparticipant_connectedr AgentsConsoleF),_proc_info_room _on_connect _on_shutdown_shutdown_callbacks_participant_entrypointsdicttupler>r _PARTICIPANT_ENTRYPOINT_CALLBACKasyncioTask_participant_taskslistr _pending_taskson_participant_available _inf_executorr^rP _log_filter_handlers_with_filter_primary_agent_sessiontempfileTemporaryDirectory_tempdirclirr get_instanceenabledsession_directory_session_directoryrr? _connectedLock_lockr_tagger_recording_initialized_early_log_handler) rVr\rhrirkrmrorrcr)r)r*rUs:       zJobContext.__init__cCs0t}|jD]}||j|j|qdSrS)logging getLoggerhandlers addFilterrrappend)rV root_loggerhandlerr)r)r* _on_setups   zJobContext._on_setupcCsB|jjs |jjjs dSt|jjsdSt|_t |jdS)z9Start buffering logs early so crash logs can be uploaded.N) rtrMrHenable_recordingrrIrrrr addHandlerrVr)r)r*_start_log_bufferings  zJobContext._start_log_bufferingcCs*|j}|dur dSt|d|_dS)z/Remove the buffering handler without replaying.N)rrr removeHandler)rVrr)r)r*_stop_log_bufferings  zJobContext._stop_log_bufferingreplayrLcCst|j}|dur dSt|d|_|sdSddlm}tjD]}t||r7|jD]}| |q,dSq"dS)zDRemove buffering handler and optionally replay records through OTLP.Nr)LoggingHandler) rrrropentelemetry.sdk._logsrr isinstancebufferemit)rVrrrhrYr)r)r*_flush_early_log_buffers    z"JobContext._flush_early_log_bufferc sddlm}|j}sdS|}||}|jrw|jrwzKtj| dd}ddl }ddl }|j j |jddIdH|j|jdd d 4IdH}||IdHWdIdHn 1IdHsbwYWntyvtd Ynwt|jjp|jj}t|js|rt|jjrz't|jjj} | st d |jjt!|jj"j#| ||jt$%d IdHWdStytdYdSwdSdS)Nrrq)indentrT)exist_okzsession_report.jsonw)modezfailed to save session reportzinvalid cloud hostname: ) agent_namecloud_hostnamereporttagger http_sessionz4failed to upload the session report to LiveKit Cloud)&rrrrrmake_session_reportrrYjsondumpsto_dictaiofiles aiofiles.ososmakedirsropenwrite Exceptionr exceptionrLr evaluationsoutcome_reasonanyrecording_optionsvaluesrrtrIr hostname ValueErrorrrHrrr) rVrrsessionrr report_jsonrf has_evalsrr)r)r*_on_session_endsN     (  zJobContext._on_session_endcCs|jdur.|js.zddlm}||dddddWnty-td|Ynw|j t |j D]}| |jq9|j dS)Nr)r!FT)audiotraceslogs transcriptz%failed to initialize crash log upload)rrvoice.agent_sessionr!init_recordingrrrrrcleanuprr removeFilterrclear)rVr!rr)r)r* _on_cleanups      zJobContext._on_cleanupcC|jjSrS)rtrMrr)r)r* is_fake_job(szJobContext.is_fake_jobrcCs t|jSrS)rrrr)r)r*r+s zJobContext.session_directorycC|jSrS)rrr)r)r*ro/zJobContext.inference_executorrcCr)asReturns the Tagger for adding tags and outcomes to the session. Tags are uploaded to LiveKit Cloud at session end. Example: ```python ctx.tagger.success(reason="Task completed successfully") ctx.tagger.fail(reason="User hung up before completing") ctx.tagger.add("voicemail:true") ``` )rrr)r)r*r3s zJobContext.taggerNrAgentSession | Noner#c Csddlm}|p |j}|std|j}|r|jrtd||j|jj|jj j |jj j |j |r2|j nd|r8|jnd|j|j|jd }|r[|j rM|j |_|jr[|j|_|j|j|_|S)Nrr"z0Cannot prepare report, no AgentSession was foundzHCannot create the AgentSession report, the RecorderIO is still recording) rjob_idroom_idrioptionsaudio_recording_pathaudio_recording_started_at started_atevents chat_history) voice.reportr#rr' _recorder_io recording_recording_optionsrHidrisidr?r output_pathrecording_started_at _started_at_recorded_eventshistorycopyrr timestampduration)rVrr# recorder_iosrr)r)r*rBs8      zJobContext.make_session_reportapi.LiveKitAPIcCstjtdS)ahReturns an LiveKitAPI for making API calls to LiveKit. Credentials are sourced from environment variables if not provided explicitly. When starting via the worker, values passed in `WorkerOptions` are exported to LIVEKIT_URL, LIVEKIT_API_KEY, and LIVEKIT_API_SECRET so this API is always usable inside job entrypoints. )r)r LiveKitAPIrrrr)r)r*r gs zJobContext.apicCr)zOReturns the process running the job. Useful for storing process-specific state.)rsrr)r)r*r\rszJobContext.procrGcCr)z5Returns the current job that the worker is executing.)rtrHrr)r)r*rHwzJobContext.jobr>cCr)zReturns the id of the worker.)rtrKrr)r)r*rK|rzJobContext.worker_idcCr)aThe Room object is the main interface that the worker should interact with. When the entrypoint is called, the worker has not connected to the Room yet. Certain properties of Room would not be available before calling JobContext.connect() )rurr)r)r*riszJobContext.roomrtc.LocalParticipantcCrrS)rulocal_participantrr)r)r*rzJobContext.agentr cCs|jstd|jS)z.Returns the primary AgentSession for this job.z(No AgentSession was started for this job)rr'rr)r)r*primary_sessionszJobContext.primary_sessioncCs|j}r |S|jjjSrS) token_claimsr@rur)rVr@r)r)r*local_participant_identitys z%JobContext.local_participant_identitydict[str, Any]cCr)a Returns the current dictionary of log fields that will be injected into log records. These fields enable enriched structured logging and can include job metadata, worker ID, trace IDs, or other diagnostic context. The returned dictionary can be directly edited, or entirely replaced via assignment (e.g., `job_context.log_context_fields = {...}`) r^rr)r)r*log_context_fieldss zJobContext.log_context_fieldsfieldscCs ||_dS)aY Sets the log fields to be injected into future log records. Args: fields (dict[str, Any]): A dictionary of key-value pairs representing structured data to attach to each log entry. Typically includes contextual information like job ID, trace information, or worker metadata. Nr)rVrr)r)r*rs callbackXCallable[[], Coroutine[None, None, None]] | Callable[[str], Coroutine[None, None, None]]csLtrdnd}jj|kr|jd Sd fdd }|j|d S) z Add a callback to be called when the job is shutting down. Optionally the callback can take a single argument, the shutdown reason. rr_r>r$rRcsIdHdSrSr)rrr)r*wrappersz1JobContext.add_shutdown_callback..wrapperN)rr>r$rR)inspectismethod__code__ co_argcountrxr)rVr min_args_numrr)rr*add_shutdown_callbacks  z JobContext.add_shutdown_callbackr@kindr@ str | NonerClist[rtc.ParticipantKind.ValueType] | rtc.ParticipantKind.ValueTypertc.RemoteParticipantcst|j||dIdHS)z Returns a participant that matches the given identity. If identity is None, the first participant that joins the room will be returned. If the participant has already joined, the function will return immediately. rN)rru)rVr@rr)r)r*rs zJobContext.wait_for_participante2eeauto_subscribe rtc_configrrtc.E2EEOptions | Nonerr4rrtc.RtcConfiguration | Nonec s|j4IdHQ|jr WdIdHdStj||tjk|d}|jj|jj |jj |dIdH| |jj D]}||q=t|j|d|_WdIdHdS1IdHs_wYdS)aConnect to the room. This method should be called only once. Args: e2ee: End-to-end encryption options. If provided, the Agent will utilize end-to-end encryption. Note: clients will also need to handle E2EE. auto_subscribe: Whether to automatically subscribe to tracks. Default is AutoSubscribe.SUBSCRIBE_ALL. rtc_config: Custom RTC configuration to use when connecting to the room. Nr)rT)rrr RoomOptionsr4r9ruconnectrtrIrJrvremote_participantsrr_apply_auto_subscribe_opts)rVrrr room_optionspr)r)r*rs"   .zJobContext.connect room_name&asyncio.Future[api.DeleteRoomResponse]csrrtdtjtj}|t|Sd fdd }t|j  fddS) z2Deletes the room and disconnects all participants.z;job_ctx.delete_room() is not executed while in console moder$rRc szjjtjp jjdIdHWdStjy&t dYdStj yO}z|j tj j krDt d|WYd}~dSWYd}~dSd}~wty]tdYdSw)N)riz'server disconnected while deleting roomzerror while deleting room: z!unknown error while deleting room)r ri delete_roomDeleteRoomRequestrur?aiohttpServerDisconnectedErrorrwarning TwirpErrorcodeTwirpErrorCode NOT_FOUNDrr)e)r rVr)r* _delete_rooms z,JobContext.delete_room.._delete_roomc jSrSrremoverrVtaskr)r*  z(JobContext.delete_room..Nr$rR) rrr&r}Futurer DeleteRoomResponse set_result create_taskrradd_done_callback)rVr futr,r))r rVr1r*r"s  zJobContext.delete_roomzSIP-participant)participant_namecall_totrunk_idparticipant_identityr;NotGivenOr[str]&asyncio.Future[api.SIPParticipantInfo]c srtdtjtj}|t|Stjj tj j j |||t|r-|nddjfddS)ah Add a SIP participant to the room. Args: call_to: The number or SIP destination to transfer the participant to. This can either be a number (+12345555555) or a sip host (sip:@) trunk_id: The ID of the SIP trunk to use participant_identity: The identity of the participant to add participant_name: The name of the participant to add Make sure you have an outbound SIP trunk created in LiveKit. See https://docs.livekit.io/sip/trunk-outbound/ for more information. zCjob_ctx.add_sip_participant() is not executed while in console modeN)r r> sip_trunk_id sip_call_tor;cr-rSr.rr0r)r*r28r3z0JobContext.add_sip_participant..)rrr&r}r5r SIPParticipantInfor7r8sipcreate_sip_participantCreateSIPParticipantRequestrur?rrrr9)rVr<r=r>r;r:r)r0r*add_sip_participants&  zJobContext.add_sip_participantF participantrtc.RemoteParticipant | str transfer_to play_dialtonec srtdtjtj}|t|St|t j r.|j t j j ks*Jd|j}n|}tjjtjjj|||djfddS)aTransfer a SIP participant to another number. Args: participant: The participant to transfer transfer_to: The number or SIP destination to transfer the participant to. This can either be a number (+12345555555) or a sip host (sip:@) play_dialtone: Whether to play a dialtone during transfer. Defaults to True. Returns: Future that completes when the transfer is complete Make sure you have enabled call transfer on your provider SIP trunk. See https://docs.livekit.io/sip/transfer-cold/ for more information. zHjob_ctx.transfer_sip_participant() is not executed while in console modez%Participant must be a SIP participant)r r>rJrKcr-rSr.rr0r)r*r2lr3z5JobContext.transfer_sip_participant..)rrr&r}r5r rCr7rr RemoteParticipantrParticipantKindPARTICIPANT_KIND_SIPr@r8rDtransfer_sip_participantTransferSIPParticipantRequestrur?rrr9)rVrHrJrKr:r>r)r0r*rO;s4  z#JobContext.transfer_sip_participantreasoncCs||dSrS)rw)rVrRr)r)r*shutdownoszJobContext.shutdown)rentrypoint_fncJCallable[[JobContext, rtc.RemoteParticipant], Coroutine[None, None, None]]rr cGs0|dd|jDvrtd|j||fdS)a!Adds an entrypoint function to be run when a participant joins the room. In cases where the participant has already joined, the entrypoint will be run immediately. Multiple unique entrypoints can be added and they will each be run in parallel for each participant. cSsg|]\}}|qSr)r)).0r+rr)r)r* ~sz9JobContext.add_participant_entrypoint..z*entrypoints cannot be added more than onceN)ryrr)rVrTrrr)r)r*add_participant_entrypointrs z%JobContext.add_participant_entrypointrr!cCs|js t|jjs|dSd|_t|jjj}tjdd|id|r>t |j j j |j j ||d|dd|jdddSdS) NTzconfiguring session recordingr)extrarr)rrr enable_traces enable_logs)r)rrrtrIrr rrdebugrrHrirrr)rVrrr)r)r*rs  zJobContext.init_recordingrcsjD]P\}}t|trj|vrqnj|krqj|fjvr+tdjddjd|j}t j ||d}|jj|f<| |ffdd qdS)Nzca participant has joined before a prior participant task matching the same identity has finished: ''z part-entry--)r?csjj|fSrS)rpopr@)rcororrVr)r*r2sz3JobContext._participant_available..) ryrrrr@rrr&r/r}r8r9)rVrr`r task_namer1r)rar*rs$    z!JobContext._participant_availablercCstj|jjddS)NF)verify_signature)r TokenVerifierverifyrtrJrr)r)r*rszJobContext.token_claims)r\rgrhrErirjrkrlrmrnrorr$rRr4)rrLr$rR)r$rL)r$r)r$r)r$rrS)rrr$r#)r$r)r$rgr$rGr$r>)r$rj)r$r)r$r )r$r)rrr$rR)rrr$rR)r@rrrr$r)rrrr4rrr$rR)r rr$r!) r<r>r=r>r>r>r;r?r$r@)F)rHrIrJr>rKrLr$r@)rQ)rRr>r$rR)rTrUrr rrr$rR)rr!r$rR)rrr$rR)r$r)/r/r0r1rr rLrr|rUrrrrrrrpropertyrrorr functoolscached_propertyr r\rHrKrirrrrsetterrrOrr4r9rr"rGrOrSrXrrrr)r)r)r*rst  3    *     %             / 4  rirjrrRcsftjtjfvr dSd fdd |jD]}|jD]}|qq|dd fd d }dS)Npubrtc.RemoteTrackPublicationr$rRcsFtjkr |jtjjkstjkr|jtjjkr!|ddSdSdSr[) r4r;rr TrackKind KIND_AUDIOr< KIND_VIDEOset_subscribed)rl)rr)r*_subscribe_if_neededsz8_apply_auto_subscribe_opts.._subscribe_if_neededtrack_publishedrrcs |dSrSr))rlr)rrr)r*on_track_publisheds z6_apply_auto_subscribe_opts..on_track_published)rlrmr$rR)rlrmrrr$rR)r4r;r<rrtrack_publicationsr)rirrrlrtr))rrrr*rs rc@s\eZdZdd d Zedd d ZedddZedddZedddZedddZ dS)rgr]r,user_arguments Any | None http_proxyrr$rRcCs&||_t|_i|_||_||_dSrS)_executor_typempcurrent_process_mp_proc _userdata_user_arguments _http_proxy)rVr]rvrxr)r)r*rUs   zJobProcess.__init__cCrrS)ryrr)r)r*r]rzJobProcess.executor_type int | NonecCrrS)r|pidrr)r)r*rrzJobProcess.piddict[Any, Any]cCrrS)r}rr)r)r*userdatarzJobProcess.userdatacCrrS)r~rr)r)r*rvrzJobProcess.user_argumentscCrrS)rrr)r)r*rxrzJobProcess.http_proxyN)r]r,rvrwrxrr$rR)r$r,)r$r)r$r)r$rw)r$r) r/r0r1rUrhr]rrrvrxr)r)r)r*rgs     rgc@seZdZd(d d Zed)d d Zed*ddZed+ddZed,ddZed)ddZ ddd-ddZ ddddd d.d&d'Z dS)/ JobRequestrHrG on_reject-Callable[[bool], Coroutine[None, None, None]] on_accept;Callable[[JobAcceptArguments], Coroutine[None, None, None]]r$rRcCs ||_t|_||_||_dSrS)_jobr}rr _on_reject _on_accept)rVrHrrr)r)r*rUs  zJobRequest.__init__r>cCrrS)rrrr)r)r*rrz JobRequest.idcCrrS)rrr)r)r*rHrzJobRequest.job models.RoomcCrrS)rrirr)r)r*rirzJobRequest.roommodels.ParticipantInfo | NonecCrrS)rrHrr)r)r* publisherrzJobRequest.publishercCrrS)rrrr)r)r*rrzJobRequest.agent_nameT) terminaterrLcs||IdHdS)zFReject the job request. The job will not be assigned to another workerN)r)rVrr)r)r*rejectszJobRequest.rejectrQNr?r@rArCr?r@rArCrBcs4|sd|j}t||||d}||IdHdS)z[Accept the job request, and start the job if the LiveKit SFU assigns the job to our worker.zagent-rN)rr=r)rVr?r@rArCrFr)r)r*accept s zJobRequest.accept)rHrGrrrrr$rRrgrf)r$r)r$r)rrLr$rR) r?r>r@r>rAr>rCrBr$rR) r/r0r1rUrhrrHrirrrrr)r)r)r*rs$      rc@seZdZUded<ded<dS)_JobShutdownInforLuser_initiatedr>rRNrNr)r)r)r*r!s  r)r$r)rirjrr4r$rR)Q __future__rr} contextvarsrir rrmultiprocessingrzrcollections.abcrr dataclassesrenumrrpathlibrtypingr r urllib.parser r$livekitr r livekit.api.access_tokenrlivekit.protocolrrlogr observabilityr telemetryrtelemetry.tracesrrrtypesrutilsrrr utils.miscr ContextVarr%ipc.inference_executorrrr r!rr#r+get_current_job_contextr,r>r4r=rErMrNPARTICIPANT_KIND_STANDARDrOrDFilterrPrrrgrrr)r)r)r*sl                1#;