hi

Thanks for reaching out, Arun. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

from datetime import datetime import json, logging from dotenv import load_dotenv load_dotenv() for lib in ('pymongo', 'llama_index'): logging.getLogger(lib).setLevel(logging.WARNING) from api import end_call from utils import DEFAULT_SYSTEM_PROMPT, WELCOME_MESSAGE from utils import get_tts_from_config, get_stt_from_config, get_all_metadata from livekit.agents import Agent, JobContext, JobProcess, WorkerOptions, cli from livekit.agents.voice import AgentSession from livekit.plugins import azure, silero, openai def prewarm(proc: JobProcess): proc.userdata.update({ "vad": silero.VAD.load(), }) class WebAgent(Agent): def __init__(self, meta_data): self.rag = meta_data["rag"] agent_info = meta_data['agent'] super().__init__( instructions=agent_info.get("system_prompt", DEFAULT_SYSTEM_PROMPT), stt=get_stt_from_config(meta_data.get('stt', {})), llm=openai.LLM(model="gpt-4.1-mini"), tts=get_tts_from_config(meta_data['tts']) ) async def on_user_turn_completed(self, turn_ctx, new_message): if not self.rag: return query = new_message.text_content if not query or len(query) < 3: return context = await self.rag.retrieve(query) turn_ctx.add_message( role="system", content=self.rag.format_context_message(context) ) async def entrypoint(ctx: JobContext): meta_data = await get_all_metadata(ctx) agent = WebAgent(meta_data) session = AgentSession( vad=ctx.proc.userdata.get("vad") ) await session.start(agent=agent, room=ctx.room) await session.say(meta_data['agent'].get('greeting', WELCOME_MESSAGE)) start_time = datetime.now() session_user = meta_data['user'] async def on_session_end(): end_call_data = { "session_id": ctx.job.id, "started_at": start_time.isoformat(), "ended_at": datetime.now().isoformat(), "transcript": json.dumps(session.history.to_dict() if hasattr(session, "history") else {"items": []}, indent=2), "created_by": session_user.get('id'), "agent_id": meta_data['agent'].get('id'), "client_id": session_user.get('client_id') } await end_call(end_call_data) ctx.add_shutdown_callback(on_session_end) def run(): cli.run_app(WorkerOptions( entrypoint_fnc=entrypoint, prewarm_fnc=prewarm, agent_name="web-agent", port=8081, num_idle_processes=1 )) if name == "__main__": run() in to this how can in ject fucntions daynamclly

Thanks — I’m following up about the code you pasted and your question: you want to inject functions dynamically into that agent flow. To help, could you clarify which type of dynamic injection you mean and how you want it to behave? A few options — please pick one or describe your own: • Load plugin functions at runtime from Python files (e.g., a plugins folder) and register them with the Agent/AgentSession. • Provide callbacks from configuration/metadata (e.g., pass function names or code in

meta_data

that should run at specific hooks like before start, on each user turn, or on session end). • Use dependency injection for testability — pass function objects into

WebAgent

or

AgentSession

at construction time. • Evaluate arbitrary code strings (e.g.,

exec

) supplied in metadata (I can note security risks if you need this). • Use an event/handler registry so external modules can subscribe to named events (e.g., "on_user_turn", "on_session_end"). Also tell me: • Which hooks do you want to support (pre-warm, on start, on_user_turn_completed, on session end, other)? • Do you control the process filesystem (can we import modules) or must everything come from `meta_data`/a remote source? • Any security constraints (no exec allowed, limited sandbox, only signed plugins)? Answering those will let me give a focused implementation pattern and a short code example that fits your constraints.

how to add tool calsl dynamclu for emapem i willne suing same agent for muoleple simallaution i will differn ttol calls

Thanks — I’m addressing how to dynamically change tool calls so you can reuse the same agent across multiple simulations with different tools. Quick clarification so I give the right pattern: • Where do your simulation-specific tools live? (a) Python code files/modules on the host, (b) function definitions sent in `meta_data`/DB/HTTP, or (c) an external tool server (MCP) you can call? • Which hooks do you want to swap tools for? (startup/prewarm, per-session entry, on each user turn, or on session end?) • Any security constraints? (e.g., no

exec

, only signed plugins, or fully trusted code) If you confirm, I’ll give a short code pattern for the best-fit option. Meanwhile, the recommended approaches (pick one after you answer the clarifying questions): • If tools are local Python modules: import them and call

await agent.update_tools([...])

before starting each simulation. • If tools come from config/meta_data: build tools with

function_tool()

dynamically and call

await agent.update_tools(tools)

inside your

entrypoint

. • If tools come from an external tool server: configure MCP with

mcp.MCPServerHTTP(...)

and let the agent load simulation-specific tools at startup. Tell me which source (a/b/c) and which hook you need and I’ll provide a compact code example. Sources: Tool definition and use | LiveKit docs

i want injsct fucntion tool calls

Thanks — I’ll show a concise example to inject function tools dynamically into your agent using

function_tool()

and

await agent.update_tools(...)

. This fits your need to reuse one agent for multiple simulations with different tool sets. 1) Define a factory to create tools dynamically:

from livekit.agents import function_tool

def make_tool(name: str, desc: str):
    async def impl(*args, **kwargs):
        # your tool logic per simulation
        return {"name": name, "args": args, "kwargs": kwargs}
    return function_tool(impl, name=name, description=desc)

2) In your entrypoint, build tools from

meta_data

and update the agent before starting the session:

async def entrypoint(ctx: JobContext):
    meta_data = await get_all_metadata(ctx)
    agent = WebAgent(meta_data)

    # Build simulation-specific tools from meta_data (example)
    tool_defs = meta_data.get("tools", [])  # e.g., [{"name":"simA_calc","desc":"..."}]
    tools = [make_tool(td["name"], td.get("desc","")) for td in tool_defs]

    # Replace agent's tools for this simulation
    await agent.update_tools(tools)

    session = AgentSession(vad=ctx.proc.userdata.get("vad"))
    await session.start(agent=agent, room=ctx.room)
    ...

3) If you need to change tools mid-session (per-turn or per-phase), call

await agent.update_tools(...)

whenever required:

# example inside an event/hook
await agent.update_tools([make_tool("phase2_op", "phase2")])

Notes and tips: • Define tool behavior in

impl

to use simulation-specific parameters from

meta_data

. • Avoid

exec

for untrusted code; prefer factories that map names to safe implementations. • You can also load tool definitions from files or an MCP server, then convert them to

function_tool

and call

update_tools

. If you want, share a sample

meta_data["tools"]

shape or one tool’s intended behavior and I’ll convert it into a ready-to-run tool factory for your agent.