@function_tool()
async def wait_for_actual_bil...
# ask-ain
numerous-shampoo-46454
09/12/2025, 2:59 PM@function_tool()
async def wait_for_actual_bill(self, record_id: str) -> str:
"""
Wait until the actual bill amount is added in MongoDB, then return it.
If the bill is missing, the agent will keep retrying silently until it exists.
"""
while True:
try:
record = await self._get_record(record_id)
if record and record.get("actual_bill") is not None:
bill = record["actual_bill"]
return str(bill)
except Exception as e:
logger.error(f"Error checking actual bill: {e}")
return "Error while checking bill"
# Sleep to avoid hammering DB / sockets
await asyncio.sleep(5)
I want during this tool call user audio input should be disabled?
t
tall-airline-16756
09/12/2025, 2:59 PMThanks for reaching out, Mubbashir. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
tall-airline-16756
09/12/2025, 3:00 PMHi there! To temporarily disable user audio input while a specific tool is running, you can simply use the
session.input.set_audio_enabled(False)session.input.set_audio_enabled(True) Copy code
session.input.set_audio_enabled(False) # Disable audio input
try:
# Perform your tool call or operation
result = await wait_for_actual_bill(record_id)
finally:
session.input.set_audio_enabled(True) # Re-enable audio inputn
numerous-shampoo-46454
09/12/2025, 3:01 PMcan we play background music during this tool execution?
t
tall-airline-16756
09/12/2025, 3:02 PMAbsolutely, you can play background music while a tool is running by using the
BackgroundAudioPlayer Copy code
from livekit.agents import BackgroundAudioPlayer, AudioConfig, BuiltinAudioClip\n\nbackground_audio = BackgroundAudioPlayer(\n ambient_sound=AudioConfig(BuiltinAudioClip.OFFICE_AMBIENCE, volume=0.8)\n)\nawait background_audio.start(room=ctx.room, agent_session=session) Copy code
# Play a custom sound file\nbackground_audio.play("/path/to/my-custom-sound.mp3")n
numerous-shampoo-46454
09/15/2025, 12:34 PMimport logging
import asyncio
import random
import string
from datetime import datetime
from livekit.agents import Agent, function_tool, get_job_context
from livekit import api, rtc
from lib.db import get_collection
from bson import ObjectId
logger = logging.getLogger("helpful_assistant")
logger.setLevel(logging.INFO)
class HelpfulAssistant(Agent):
def __init__(self, instructions: str = "You are a helpful assistant.") -> None:
super().__init__(instructions=instructions)
self._state = {
"record_id": None,
"info_saved": False,
"promotion_done": False,
"auth_generated": False
}
self.collection = get_collection("crm")
# ---------- Utility Methods ----------
async def _get_record(self, record_id: str):
return self.collection.find_one({"_id": ObjectId(record_id)})
async def _update_record(self, record_id: str, data: dict):
return self.collection.update_one({"_id": ObjectId(record_id)}, {"$set": data})
def _clean_phone(self, phone: str) -> str:
return ''.join(filter(str.isdigit, phone))
# ---------- Agent Functions ----------
@function_tool()
async def end_call(self):
"""Gracefully end the call and close the room."""
logger.info("Ending call...")
try:
job_ctx = get_job_context()
await job_ctx.api.room.delete_room(api.DeleteRoomRequest(room=job_ctx.room.name))
except Exception as e:
logger.error(f"Error ending call: {e}")
@function_tool()
async def save_initial_customer_info(self, name: str, zip_code: str = "") -> str:
"""Store basic customer info when the conversation starts."""
caller_phone = await self._extract_caller_phone()
if not caller_phone:
logger.warning("Could not extract caller phone from SIP metadata, using dummy phone")
caller_phone = "0000000000"
# Generate serial number
serial_number = self.collection.count_documents({}) + 1
crm_data = {
"name": name,
"phone": caller_phone,
"email": "test@gmail.com",
"serial_number": serial_number,
"zip_code": zip_code,
"created_at": datetime.utcnow(),
"partial_record": True,
"actual_bill": None
}
try:
result = self.collection.insert_one(crm_data)
self._state.update({"record_id": str(result.inserted_id), "info_saved": True})
return str(result.inserted_id)
except Exception as e:
logger.error(f"Error saving info: {e}")
return f"Error: {str(e)}"
async def _extract_caller_phone(self):
"""Try to extract phone from participants or room name."""
try:
job_ctx = get_job_context()
for p in job_ctx.room.remote_participants.values():
if hasattr(p, 'kind') and p.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP:
# Use the correct attribute for phone number
if hasattr(p, 'attributes') and 'sip.phoneNumber' in p.attributes:
return p.attributes['sip.phoneNumber']
# Fallback to identity if it looks like a phone number
if p.identity.startswith("+"):
return p.identity
# Existing room name fallback
if "_" in job_ctx.room.name:
parts = job_ctx.room.name.split("_", 1)
if len(parts) > 1 and parts[1].startswith("+"):
return parts[1]
except Exception as e:
logger.warning(f"Phone extraction failed: {e}")
return None
@function_tool()
async def complete_promotion_record(self, record_id: str, bill_amount: float,
billing_department_phone: str, months: int = 6) -> str:
"""Finalize promotion data for the customer record."""
record = await self._get_record(record_id)
if not record:
return "Error: Customer record not found"
update = {
"bill_amount": bill_amount,
"months": months,
"billing_department_phone": billing_department_phone,
"card_numbers": [],
"paid_flag": False,
"sales_flag": True,
"partial_record": False,
"updated_at": datetime.utcnow(),
}
result = await self._update_record(record_id, update)
if result.modified_count:
self._state["promotion_done"] = True
return "Promotion record completed successfully"
return "Error: Failed to update promotion record"
@function_tool()
async def generate_auth_code(self, record_id: str) -> str:
"""Generate and save a random 5-digit authorization code."""
if not self._state["info_saved"]:
return "Please provide your name and phone number first."
auth_code = ''.join(random.choices(string.digits, k=5))
try:
result = await self._update_record(record_id, {
"auth_code": auth_code,
"auth_code_generated_at": datetime.utcnow()
})
if result.modified_count:
self._state["auth_generated"] = True
except Exception as e:
logger.error(f"Auth code save failed: {e}")
return auth_code
@function_tool()
async def save_payment_time(self, record_id: str, payment_time: str) -> str:
"""Save customer payment time to CRM record."""
record = await self._get_record(record_id)
if not record:
return "Error: Customer record not found"
result = await self._update_record(record_id, {
"payment_time": payment_time,
"payment_time_saved_at": datetime.utcnow()
})
return f"Payment time '{payment_time}' saved." if result.modified_count else "Error: Could not save payment time"
@function_tool()
async def wait_for_actual_bill(self, record_id: str) -> str:
"""
Wait until the actual bill amount is added in MongoDB, then return it.
If the bill is missing, the agent will keep retrying silently until it exists.
"""
while True:
try:
record = await self._get_record(record_id)
if record and record.get("actual_bill") is not None:
bill = record["actual_bill"]
return str(bill)
except Exception as e:
logger.error(f"Error checking actual bill: {e}")
return "Error while checking bill"
# Sleep to avoid hammering DB / sockets
await asyncio.sleep(5)
Please write a fuction tool for background music play which when called play background music
t
tall-airline-16756
09/15/2025, 12:34 PMI'd love to share an example of a function tool that plays background music when called. This might be just what you need to enhance your session:
Copy code
from livekit.agents import function_tool, BackgroundAudioPlayer, AudioConfig, BuiltinAudioClip
class HelpfulAssistant(Agent):
def __init__(self, instructions: str = "You are a helpful assistant.") -> None:
super().__init__(instructions=instructions)
self.background_audio = BackgroundAudioPlayer(
ambient_sound=AudioConfig(BuiltinAudioClip.OFFICE_AMBIENCE, volume=0.8)
)
@function_tool()
async def play_background_music(self):
"""Play background music during the session."""
try:
job_ctx = get_job_context()
await self.background_audio.start(room=job_ctx.room, agent_session=self)
return "Background music started successfully."
except Exception as e:
logger.error(f"Error starting background music: {e}")
return f"Error: {str(e)}"BackgroundAudioPlayerBuiltinAudioClip.OFFICE_AMBIENCE