early-butcher-66889
12/17/2025, 9:06 AMtall-airline-16756
12/17/2025, 9:06 AMtall-airline-16756
12/17/2025, 9:06 AMuv run main.py console --log-level=DEBUG
◦ Also set: LIVEKIT_EVALS_VERBOSE=1 if using evaluation tooling:
LIVEKIT_EVALS_VERBOSE=1 uv run main.py console --log-level=DEBUG This will increase the agent/server verbosity and often shows tool invocation and errors.
1. Add explicit logging inside your tool functions:
◦ Use print() or a structured logger at the start, before/after external calls, and in exception handlers. Example:
print(f"[DEBUG] Entering my_tool with params={params}")
try:
r = await httpx.get("<http://localhost:8000/...>")
print("[DEBUG] API response:", r.text)
return r.json()
except Exception as e:
print("[ERROR] API call failed:", e)
return {"error": str(e)}
• If these logs don’t appear, the function likely isn’t being invoked.
1. Wrap tool calls in try/except and return clear error strings:
◦ Returning explicit error information ensures the agent or LLM can surface the failure instead of hanging silently.
1. Verify tool definitions and schema:
◦ Ensure @function_tool() docstrings, parameter types, and return expectations are precise so the LLM routes to the tool. Ambiguous or incorrect definitions can cause non-invocation.
1. Confirm network reachability from the agent process:
◦ If the agent runs in Docker or another container, localhost may not point to your host. Try host.docker.internal or the container’s network address.
◦ Test connectivity with a small script or an httpx request from the same runtime.
1. If the agent hangs (no logs after invocation attempt):
◦ The LLM may be waiting for a long-running call or stuck before entering the tool. Increase timeouts and add early-entry logs to detect whether the tool function is hit.
1. If you need conversational visibility:
◦ Agent logs are not in the WebRTC audio stream. To expose tool-call results into the conversation, either send the results over a data track/RPC to the client UI or ensure the agent returns the tool output in the conversation transcript it emits.
If you’d like, tell me:
• The exact tool decorator/definition you’re using (brief snippet),
• Whether the agent runs in Docker, and
• A sample of one failing tool call and how you call your local API.
I can then point out specific spots to add logs or identify likely causes.
Sources: Agent server options | LiveKit docs | Testing and evaluation | LiveKit docs | Agents and handoffs | LiveKit docs | Tool definition and use | LiveKit docs | Job lifecycle | LiveKit docsearly-butcher-66889
12/17/2025, 9:30 AMasync def process_order(context: RunContext, stamp_quantity: int, envelope_quantity: int, email: str) -> str:
"""Process the customer's order for stamps and envelopes. Requires quantities and email address."""
if stamp_quantity < 0 or envelope_quantity < 0:
return "I'm sorry, quantities must be positive numbers. Please provide valid quantities."
if not email or "@" not in email:
return "I'm sorry, please provide a valid email address."
stamp_price = 0.68
envelope_price = 0.15
total = (stamp_quantity * stamp_price) + (envelope_quantity * envelope_price)
response = f"Thank you! I've received your order:\n"
response += f"- {stamp_quantity} postage stamp(s) at $0.68 each\n"
response += f"- {envelope_quantity} envelope(s) at $0.15 each\n"
response += f"Total: ${total:.2f}\n\n"
response += f"Payment and delivery details will be sent to {email}. Is there anything else I can help you with?"
return response
And then what happens is that I have a registry
TOOL_REGISTRY: dict[str, Callable] = {
"verify_customer": verify_customer,
"schedule_redelivery": schedule_redelivery,
"process_order": process_order,
"end_call": end_call,
}
And then afterwards based on the agent I then iterate and append them to the agent based on their needs
for agent_id, agent_config in agents_config.items():
agent = created_agents[agent_id]
for tool_name in agent_config.get("tools", []):
if tool_name not in tool_registry:
raise ValueError(f"Tool '{tool_name}' not found in tool registry for agent '{agent_id}'")
tool_func = tool_registry[tool_name]
# Wrap with function_tool decorator and attach to agent
decorated_tool = function_tool()(tool_func)
agent._tools.append(decorated_tool)
In my non-dynamic version (one which I hard coded without this pattern) I can see the log line tools execution completed {"speech_id": "speech_7d79aed6e3f7"} but in the one I try to generate it doesn't seem to be shown.
Why is that? I use the same pattern to do transfers and the log line appearstall-airline-16756
12/17/2025, 9:30 AMfunction_tool()(tool_func) can produce a wrapper that isn’t fully registered. Create the tool with explicit metadata:
dynamic_tool = function_tool(
name=tool_name,
description=tool_func.__doc__.strip() or "No description",
)(tool_func)
1. Don’t mutate agent._tools directly — use the agent API
◦ Appending to agent._tools bypasses internal registration, observability, and event emission. Register tools with:
await agent.update_tools(agent.tools + [dynamic_tool])
• This ensures the framework emits the tool-executed events that produce the "tools execution completed" log.
Quick debugging checks
• After registering, log agent.tools (or inspect agent.tools) to confirm the new tool appears with the correct name and callable.
• Run with debug logs: uv run main.py console --log-level=DEBUG (and LIVEKIT_EVALS_VERBOSE=1 if relevant).
• Add early `print()`/logger statements at the top of the tool function to confirm it’s actually entered.
If you share a minimal snippet of your final dynamic-registration loop (after you apply the name/description and update_tools()), I can point out any remaining issues.
Sources: Tool definition and use | LiveKit docs | Events and error handling | LiveKit docs | Testing and evaluation | LiveKit docs | Log collection | LiveKit Docsearly-butcher-66889
12/17/2025, 9:46 AM# Step 2: Attach functional tools to each agent
for agent_id, agent_config in agents_config.items():
agent = created_agents[agent_id]
for tool_name in agent_config.get("tools", []):
if tool_name not in tool_registry:
raise ValueError(f"Tool '{tool_name}' not found in tool registry for agent '{agent_id}'")
tool_func = tool_registry[tool_name]
# Wrap with function_tool decorator and attach to agent
decorated_tool = function_tool()(tool_func)
await agent.update_tools(agent.tools + [decorated_tool])
# Step 3: Create and attach transfer tools
for agent_id, agent_config in agents_config.items():
agent = created_agents[agent_id]
for transfer in agent_config.get("transfers_to", []):
target_agent_id = transfer["agent_id"]
if target_agent_id not in created_agents:
raise ValueError(f"Transfer target '{target_agent_id}' not found for agent '{agent_id}'")
target_agent = created_agents[target_agent_id]
transfer_func = create_transfer_tool(target_agent, target_agent_id)
# Wrap with function_tool decorator and attach to agent
decorated_transfer = function_tool()(transfer_func)
await agent.update_tools(agent.tools + [decorated_transfer])tall-airline-16756
12/17/2025, 9:47 AMfunction_tool()(tool_func) and calling await agent.update_tools(agent.tools + [decorated_tool]) (good you used update_tools, but the decorator usage is the likely issue).
Why this is happening (short)
• Tools appear as plain function objects in your logs because function_tool() was invoked without explicit metadata; that can result in wrappers the agent/LLM doesn’t fully register for tool calling/observability. When a tool isn’t correctly registered the agent may think it invoked a tool (LLM-side) but the runtime never emits the execution events or enters your function (so no logs).
Concrete fixes
1. Create the tool with name and description when constructing dynamically:
tool_description = tool_func.__doc__.strip() if tool_func.__doc__ else "No description"
dynamic_tool = function_tool(
name=tool_name,
description=tool_description,
)(tool_func)
await agent.update_tools(agent.tools + [dynamic_tool])
1. Do the same for transfer tools:
decorated_transfer = function_tool(
name=f"transfer_to_{target_agent_id}",
description=transfer_func.__doc__ or f"Transfer to {target_agent_id}",
)(transfer_func)
await agent.update_tools(agent.tools + [decorated_transfer])
1. Add early-entry logging in each tool to confirm it’s executed:
print("[TOOL] entering process_order", locals())
1. After update, verify registration:
print("Registered tools:", [t.name for t in agent.tools])
(ensure t.name exists — if not, that indicates improper tool creation)
Extra debugging steps
• Run with debug logs: uv run main.py console --log-level=DEBUG and LIVEKIT_EVALS_VERBOSE=1.
• Confirm the LLM received the tool schema (check the request payloads or debug output showing tools sent to the model).
• If running inside containers, ensure no networking/timeout issues are preventing tool completion.
If you’d like, share:
• One full example of how you create a single dynamic tool (the function and the exact decorated object), and
• The debug log lines showing the LLM/tool negotiation (if available).
I’ll point out the exact spot that still needs changing.early-butcher-66889
12/17/2025, 10:00 AMfrom dotenv import load_dotenv
from typing import Optional, Callable, Any, TypedDict
from livekit import agents, rtc
from livekit.agents import AgentServer, AgentSession, Agent, ChatContext, room_io, function_tool, RunContext
from livekit.plugins import noise_cancellation, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel
load_dotenv(".env.local")
# =============================================================================
# TOOL IMPLEMENTATIONS
# =============================================================================
async def verify_customer(context: RunContext, name: str, date_of_birth: str) -> str:
"""Verify the customer's identity using their name and date of birth. Format: YYYY-MM-DD"""
return f"Thank you {name}. I've verified your identity. You have 2 packages: PKG-001 (held at post office) and PKG-002 (in transit)."
async def process_order(context: RunContext, stamp_quantity: int, email: str) -> str:
"""Process the customer's order for stamps. Requires quantity and email address."""
if stamp_quantity < 0:
return "I'm sorry, quantity must be a positive number."
if not email or "@" not in email:
return "I'm sorry, please provide a valid email address."
total = stamp_quantity * 0.68
return f"Thank you! I've received your order for {stamp_quantity} stamps. Total: ${total:.2f}. Details will be sent to {email}."
# =============================================================================
# TOOL REGISTRY
# =============================================================================
class ToolEntry(TypedDict):
"""Type definition for a tool entry with function, description, and name."""
function: Callable[..., Any]
description: str
name: str
TOOL_REGISTRY: dict[str, ToolEntry] = {
"verify_customer": {
"function": verify_customer,
"description": "Verify the customer's identity using their name and date of birth. Format: YYYY-MM-DD",
"name": "verify_customer"
},
"process_order": {
"function": process_order,
"description": "Process the customer's order for stamps. Requires quantity and email address.",
"name": "process_order"
}
}
# =============================================================================
# AGENT CONFIG
# =============================================================================
AGENT_CONFIG = {
"initial_agent": "main",
"agents": {
"main": {
"instructions": "You are a Post Office assistant. Ask what the customer needs help with: deliveries or ordering stamps.",
"on_enter": "Greet the user and ask how you can help them today.",
"tools": [],
"transfers_to": [
{
"agent_id": "delivery",
"description": "Transfer to delivery specialist for package inquiries."
},
{
"agent_id": "stamps",
"description": "Transfer to stamps specialist for ordering stamps."
}
]
},
"delivery": {
"instructions": "You are a delivery specialist. Verify the customer's identity by asking for their name and date of birth, then use the verify_customer tool.",
"on_enter": "Introduce yourself as the delivery specialist and ask for their name and date of birth.",
"tools": ["verify_customer"],
"transfers_to": []
},
"stamps": {
"instructions": "You are a stamps specialist. Help customers order stamps at $0.68 each. Ask for quantity and email, then use the process_order tool.",
"on_enter": "Introduce yourself as the stamps specialist. Stamps are $0.68 each. Ask what quantity they need and their email.",
"tools": ["process_order"],
"transfers_to": []
}
}
}
# =============================================================================
# DYNAMIC AGENT FACTORY
# =============================================================================
class DynamicAgent(Agent):
"""A dynamically configured agent that can be created from a config dict."""
def __init__(
self,
instructions: str,
on_enter_instructions: Optional[str] = None,
chat_ctx: Optional[ChatContext] = None,
) -> None:
super().__init__(
instructions=instructions,
chat_ctx=chat_ctx,
)
self._on_enter_instructions = on_enter_instructions
async def on_enter(self) -> None:
await self.session.generate_reply(
instructions=self._on_enter_instructions
)
def create_transfer_tool(target_agent: Agent, agent_id: str) -> Callable:
"""Create a transfer tool function that returns the target agent."""
async def transfer_tool(context: RunContext) -> Agent:
# Create a new instance with copied chat history
return DynamicAgent(
instructions=target_agent._instructions,
on_enter_instructions=target_agent._on_enter_instructions,
chat_ctx=context.session.history.copy(),
)
# Set the function name and docstring
transfer_tool.__name__ = f"transfer_to_{agent_id}"
transfer_tool.__doc__ = f"Transfer the customer to the {agent_id} agent."
return transfer_tool
async def build_agents_from_config(
config: dict,
tool_registry: dict[str, ToolEntry],
) -> Agent:
"""Build all agents from config and return the initial agent."""
agents_config = config["agents"]
initial_agent_id = config["initial_agent"]
# Step 1: Create all agents
created_agents: dict[str, DynamicAgent] = {}
for agent_id, agent_config in agents_config.items():
agent = DynamicAgent(
instructions=agent_config["instructions"],
on_enter_instructions=agent_config.get("on_enter"),
)
created_agents[agent_id] = agent
print(f"Created agent '{agent_id}'")
# Step 2: Attach functional tools to each agent
for agent_id, agent_config in agents_config.items():
agent = created_agents[agent_id]
for tool_name in agent_config.get("tools", []):
if tool_name not in tool_registry:
raise ValueError(f"Tool '{tool_name}' not found in tool registry")
tool_entry = tool_registry[tool_name]
decorated_tool = function_tool(
name=tool_entry["name"],
description=tool_entry["description"],
)(tool_entry["function"])
await agent.update_tools(agent.tools + [decorated_tool])
print(f" Added tool '{tool_name}' to agent '{agent_id}'")
# Step 3: Create and attach transfer tools
for agent_id, agent_config in agents_config.items():
agent = created_agents[agent_id]
for transfer in agent_config.get("transfers_to", []):
target_agent_id = transfer["agent_id"]
if target_agent_id not in created_agents:
raise ValueError(f"Transfer target '{target_agent_id}' not found")
target_agent = created_agents[target_agent_id]
transfer_func = create_transfer_tool(target_agent, target_agent_id)
decorated_transfer = function_tool()(transfer_func)
await agent.update_tools(agent.tools + [decorated_transfer])
print(f" Added transfer 'transfer_to_{target_agent_id}' to agent '{agent_id}'")
# Audit all agents
print("\n=== AGENT AUDIT ===")
for agent_id, agent in created_agents.items():
print(f"\nAgent '{agent_id}' has {len(agent.tools)} tools:")
for tool in agent.tools:
print(f" - {tool.__name__}: {tool.__doc__}")
# Return the initial agent
if initial_agent_id not in created_agents:
raise ValueError(f"Initial agent '{initial_agent_id}' not found in config")
return created_agents[initial_agent_id]
# =============================================================================
# SERVER SETUP
# =============================================================================
server = AgentServer()
@server.rtc_session()
async def my_agent(ctx: agents.JobContext):
# Build agents from config
initial_agent = await build_agents_from_config(AGENT_CONFIG, TOOL_REGISTRY)
session = AgentSession(
stt="assemblyai/universal-streaming:en",
llm="openai/gpt-4.1-mini",
tts="cartesia/sonic-3:9626c31c-bec5-4cca-baa8-f8ba9e84c8bc",
vad=silero.VAD.load(),
turn_detection=MultilingualModel(),
)
await session.start(
room=ctx.room,
agent=initial_agent,
room_options=room_io.RoomOptions(
audio_input=room_io.AudioInputOptions(
noise_cancellation=lambda params: noise_cancellation.BVCTelephony() if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP else noise_cancellation.BVC(),
),
),
)
if __name__ == "__main__":
agents.cli.run_app(server)tall-airline-16756
12/17/2025, 10:00 AMfunction_tool() without name/description, and you also repeatedly call await agent.update_tools(agent.tools + [decorated_tool]) inside the loops which can cause race/overwrite issues if not done carefully.
What to change (concise)
1) Always create tools with explicit name + description
• For registry tools you already do this — keep it.
• For transfer tools do the same instead of `function_tool()(transfer_func)`:
decorated_transfer = function_tool(
name=f"transfer_to_{target_agent_id}",
description=transfer_func.__doc__ or f"Transfer to {target_agent_id}",
)(transfer_func)
await agent.update_tools(agent.tools + [decorated_transfer])
2) Avoid repeatedly calling update_tools in a tight loop without collecting tools first
• Collect tools to append, then call update once per agent to avoid intermediate inconsistent states:
new_tools = []
for tool_name in agent_config.get("tools", []):
entry = tool_registry[tool_name]
new_tools.append(function_tool(name=entry["name"], description=entry["description"])(entry["function"]))
for transfer in agent_config.get("transfers_to", []):
tf = create_transfer_tool(created_agents[transfer["agent_id"]], transfer["agent_id"])
new_tools.append(function_tool(name=f"transfer_to_{transfer['agent_id']}", description=tf.__doc__)(tf))
if new_tools:
await agent.update_tools(agent.tools + new_tools)
print(f" Added {len(new_tools)} tools to agent '{agent_id}'")
3) Add early-entry logs inside tools and verify registration
• Put a print/log at the top of each tool:
print(f"[TOOL] entering {process_order.__name__} with args: ...")
• After update, confirm:
print("Registered tools:", [(t.name if hasattr(t, 'name') else getattr(t, '__name__', str(t))) for t in agent.tools])
If the tool objects lack .name, that indicates they were not created with function_tool(name=...).
4) Run with verbose/debug logs
• Start server with:
LIVEKIT_EVALS_VERBOSE=1 uv run main.py console --log-level=DEBUG This shows model/tool negotiation and function_tools_executed events.
Why this fixes it
• function_tool() without metadata can produce wrappers the agent/LLM doesn’t expose to the model schema or observability pipeline, so the model may attempt a function call but the runtime never emits execution events or enters the function. Explicit names/descriptions + updating tools in one call ensure the tools are registered and visible to the LLM, which produces the "tools execution completed" log.
If you want I can:
• Provide a minimal refactor of your build loop with the fixes applied, or
• Inspect the output of the tool-registration print statements you get after these changes. Which would you prefer?early-butcher-66889
12/17/2025, 10:10 AMtall-airline-16756
12/17/2025, 10:11 AM--log-level.
Short answers
• The agent hanging when you reference a tool usually means the model attempted a function call but the runtime never executed it (tool schema mismatch, missing metadata, or the tool returned an unexpected type).
• --log-level isn’t recognized because your current CLI invocation doesn’t accept that flag. Use environment variables or the server CLI start subcommand to enable debug logging.
What to try next (concise troubleshooting steps)
1) Fix transfer tools (must include name + description)
• Replace your current transfer wrapping:
decorated_transfer = function_tool()(transfer_func)
with:
decorated_transfer = function_tool(
name=f"transfer_to_{target_agent_id}",
description=transfer_func.__doc__ or f"Transfer to {target_agent_id}",
)(transfer_func)
This ensures the tool schema is sent to the LLM and the runtime tracks execution.
2) Batch tool updates per agent (avoid repeated update calls)
• Collect all new tools, then call update once:
new_tools = [...]
if new_tools:
await agent.update_tools(agent.tools + new_tools)
Multiple sequential updates can create transient states the model doesn’t see.
3) Confirm tool signatures and return types
• Tools must accept the expected parameters and return acceptable types:
◦ Transfer tools that return an Agent should return a proper Agent instance (your DynamicAgent is fine), not None or an unexpected object.
◦ Other tools should return strings or serializable dicts. If the model receives something unexpected it can hang.
4) Add early-entry logs inside every tool
• Add a print/log at the top of each tool to confirm execution:
print("[TOOL] entering process_order", {"args": ..., "context_session_id": id(context.session)})
If you never see that line, the function was not invoked.
5) Verify the model received the tool schema
• After update, print agent.tools to confirm tools have .name and `.__doc__`:
print("Registered tools:", [(getattr(t, "name", None), getattr(t, "__doc__", None)) for t in agent.tools])
If name is missing, the tool wasn’t created correctly.
6) Enable debug logging correctly
• Option A — use the server CLI start subcommand if available:
uv run main.py start --log-level=DEBUG (Only valid if your app exposes a start command.)
• Option B — set env vars before running the console command:
LIVEKIT_LOG_LEVEL=DEBUG LIVEKIT_EVALS_VERBOSE=1 uv run main.py console
• Option C — set logging in code at startup if CLI flags aren’t available:
import logging
logging.basicConfig(level=logging.DEBUG)
Then run uv run main.py console.
7) If the model still hangs when you explicitly reference a tool in the prompt
• Capture the model/tool negotiation in logs (with debug enabled) and confirm:
◦ The model’s tool call JSON matches the tool name and parameters.
◦ The runtime receives that call and enters the tool (check your early-entry prints).
• If the model calls the right tool but you don’t see entry logs, the call may be blocked by awaiting something else or a coroutine not scheduled — ensure your tool is async and uses awaited IO properly.
If you want, I can:
• Provide a minimal, refactored version of your build loop (with batched updates and transfer-tool creation) you can drop into your file, or
• Inspect the output you get after enabling logging and adding the early-entry prints.
Which would you like?
Sources: Agent server options | LiveKit docs | Tool definition and use | LiveKit docsearly-butcher-66889
12/17/2025, 10:13 AMtall-airline-16756
12/17/2025, 10:14 AMfunction_tool(name=..., description=...)(func) so the model receives a usable schema. If the transferred agent lacks tools with matching names, the runtime won’t enter your async function (no logs).
Immediate checks
• After a transfer, do you see the transferred agent’s tool list in the logs? Add a print at transfer return:
print("Transferred agent tools:", [(getattr(t,'name',None), getattr(t,'*doc*',None)) for t in new_agent.tools])
• Add early-entry prints at the top of transfer and target tools to confirm whether the function is entered.
• Confirm the LLM tool call payload in debug logs: it must reference the exact name you registered.
Two safe fixes 1) Attach tools to the DynamicAgent when you create it in the transfer tool (so the new agent already has tools):
• Build decorated tools (with name+description) and call await new_agent.update_tools(new_agent.tools + new_tools) before returning the agent from the transfer function.
2) Return an agent that references an existing pre-built agent object that already has tools registered (avoid returning a bare agent with no tools).
Minimal pattern to preserve tools on transfer
• When creating the transfer function, copy tools from the target agent (the agent you created earlier, which already has tools registered) into the new DynamicAgent before return:
def create_transfer_tool(target_agent: Agent, agent_id: str) -> Callable:
async def transfer_tool(context: RunContext) -> Agent:
new_agent = DynamicAgent(
instructions=target_agent._instructions,
on_enter_instructions=target_agent._on_enter_instructions,
chat_ctx=context.session.history.copy(),
)
# Copy and attach tools from the original target_agent (preserving name/description)
# Recreate decorated tools from the target_agent.tools entries if needed,
# or use target_agent.tools directly if they are proper FunctionTool objects.
existing_tools = []
for t in target_agent.tools:
# If t already has metadata and is a proper object, reuse it; otherwise recreate
if hasattr(t, "name") and hasattr(t, "__doc__"):
existing_tools.append(t)
else:
# fallback: reconstruct from underlying function
existing_tools.append(function_tool(
name=getattr(t, "__name__", f"tool_{id(t)}"),
description=getattr(t, "__doc__", "") or "No description"
)(getattr(t, "__wrapped__", t)))
if existing_tools:
await new_agent.update_tools(new_agent.tools + existing_tools)
return new_agent
transfer_tool.__name__ = f"transfer_to_{agent_id}"
transfer_tool.__doc__ = f"Transfer the customer to the {agent_id} agent."
return transfer_tool
Batch attach tools when building agents (reminder)
• When you first build agents, collect all tools and transfer-tools per agent and call await agent.update_tools(agent.tools + new_tools) once per agent (avoid many sequential updates).
Debugging steps to collect useful logs
• Add early-entry prints at top of transfer_tool and every target tool.
• Enable debug logging (if CLI flag not available, set logging in code):
import logging
logging.basicConfig(level=logging.DEBUG)
• After a transfer occurs, log the new agent.tools right away to confirm names/schemas.
About LiveKit examples
• The LiveKit team’s examples demonstrate dynamic tool registration patterns and using update_tools() — follow the pattern above: ensure tools are given name and description and that agents returned by transfers already have those tools registered. If you want, I can produce a minimal, runnable refactor of your file that:
◦ Builds agents
◦ Registers tools (including transfer tools) with metadata
◦ Ensures transferred agents have tools attached before return
Would you like that minimal runnable refactor? If so, tell me which approach you prefer for transfers:
• Return a fresh agent instance with copied tools (the pattern above), or
• Return a reference to the existing pre-built agent (no copy), and I’ll produce the snippet.early-butcher-66889
12/17/2025, 10:14 AMtall-airline-16756
12/17/2025, 10:15 AMimport asyncio
from typing import Callable, Dict, Any
from dataclasses import dataclass
from livekit.agents import Agent, AgentServer, AgentSession, function_tool, RunContext
# --- Example tools (plain implementations) ---
async def verify_customer_impl(context: RunContext, name: str, date_of_birth: str) -> str:
print("[TOOL] verify_customer entered", name, date_of_birth)
return f"Verified {name} (DOB {date_of_birth})"
async def process_order_impl(context: RunContext, quantity: int, email: str) -> str:
print("[TOOL] process_order entered", quantity, email)
return f"Order received: {quantity} stamps -> receipt sent to {email}"
# --- Tool registry with metadata ---
TOOL_REGISTRY: Dict[str, Dict[str, Any]] = {
"verify_customer": {
"function": verify_customer_impl,
"name": "verify_customer",
"description": "Verify customer by name and DOB (YYYY-MM-DD).",
},
"process_order": {
"function": process_order_impl,
"name": "process_order",
"description": "Process stamp order: quantity and email.",
},
}
# --- Dynamic agent subclass ---
class DynamicAgent(Agent):
def __init__(self, instructions: str, on_enter: str | None = None):
super().__init__(instructions=instructions)
self._on_enter_instructions = on_enter
async def on_enter(self) -> None:
if self._on_enter_instructions:
await self.session.generate_reply(instructions=self._on_enter_instructions)
# --- Helper to create decorated tools ---
def make_decorated_tool(entry: Dict[str, Any]) -> Callable:
return function_tool(name=entry["name"], description=entry["description"])(entry["function"])
# --- Transfer tool factory: returns a new agent instance with tools already attached ---
def create_transfer_tool(target_agent: DynamicAgent, target_id: str) -> Callable:
async def transfer_tool(context: RunContext) -> DynamicAgent:
print(f"[TRANSFER] creating transferred agent for '{target_id}'")
# Create a new agent with same instructions and on_enter
new_agent = DynamicAgent(instructions=target_agent._instructions, on_enter=target_agent._on_enter_instructions)
# Copy tools from the original target_agent (recreate decorated tools to ensure metadata)
new_tools = []
for t in target_agent.tools:
# If t already has a 'name' attribute and callable, reuse; otherwise reconstruct from registry
name = getattr(t, "name", None) or getattr(t, "__name__", None)
doc = getattr(t, "__doc__", "") or ""
func = getattr(t, "__wrapped__", t)
if callable(func) and name:
new_tools.append(function_tool(name=name, description=doc)(func))
if new_tools:
# Attach all tools in batch
await new_agent.update_tools(new_agent.tools + new_tools)
print("[TRANSFER] attached tools to transferred agent:", [getattr(t, "name", None) for t in new_agent.tools])
return new_agent
transfer_tool.__name__ = f"transfer_to_{target_id}"
transfer_tool.__doc__ = f"Transfer the customer to the {target_id} agent."
return transfer_tool
# --- Build agents from a simple config ---
async def build_agents() -> DynamicAgent:
# simple config
config = {
"main": {
"instructions": "You are a front-desk assistant. Ask what user needs.",
"on_enter": "Hello — how can I help? (deliveries or stamps)",
"tools": [],
"transfers_to": ["delivery", "stamps"],
},
"delivery": {
"instructions": "You are delivery specialist. Verify user then help.",
"on_enter": "Delivery specialist here — please provide name and DOB.",
"tools": ["verify_customer"],
"transfers_to": [],
},
"stamps": {
"instructions": "You are stamps specialist. Ask quantity and email.",
"on_enter": "Stamps specialist — how many stamps do you need?",
"tools": ["process_order"],
"transfers_to": [],
},
}
# create agent instances
created: Dict[str, DynamicAgent] = {}
for aid, aconf in config.items():
created[aid] = DynamicAgent(instructions=aconf["instructions"], on_enter=aconf.get("on_enter"))
# attach tools (batched) for each agent
for aid, aconf in config.items():
agent = created[aid]
new_tools = []
for tname in aconf.get("tools", []):
entry = TOOL_REGISTRY.get(tname)
if not entry:
raise ValueError("Missing tool: " + tname)
new_tools.append(make_decorated_tool(entry))
# create transfer tools too (these will return agents that already have their tools)
for target in aconf.get("transfers_to", []):
target_agent = created[target]
transfer_fn = create_transfer_tool(target_agent, target)
# ensure decorated transfer tool has metadata
new_tools.append(function_tool(name=f"transfer_to_{target}", description=transfer_fn.__doc__)(transfer_fn))
if new_tools:
await agent.update_tools(agent.tools + new_tools)
print(f"[BUILD] Agent '{aid}' tools: {[getattr(t,'name', getattr(t,'__name__', None)) for t in agent.tools]}")
return created["main"]
# --- Minimal run harness that starts a session with the initial agent ---
async def main():
initial_agent = await build_agents()
# Create a minimal AgentSession (uses defaults; adapt to your setup)
session = AgentSession(
stt=None, # provide real STT/LLM/TTS in your app
llm=None,
tts=None,
)
# The session.start call requires a room; here we only simulate calling on_enter to validate behavior
# Simulate agent entering (which may call on_enter and show tools are present)
class DummySession:
async def generate_reply(self, instructions: str):
print("[SESSION] generate_reply:", instructions)
# Attach a dummy session to the agent so on_enter can run without full LiveKit session
initial_agent.session = DummySession()
await initial_agent.on_enter()
# Simulate a transfer call: call the decorated transfer tool directly
# Find a transfer tool on the main agent and invoke it
for t in initial_agent.tools:
name = getattr(t, "name", None) or getattr(t, "__name__", None)
if name and name.startswith("transfer_to_"):
print("[SIM] invoking", name)
# call underlying function (tools are wrapped by function_tool; unwrap if needed)
fn = getattr(t, "__wrapped__", t)
new_agent = await fn(RunContext(session=initial_agent.session)) # provide RunContext-like object
# simulate attach session and run on_enter on transferred agent
new_agent.session = DummySession()
await new_agent.on_enter()
print("[SIM] transferred agent tools:", [getattr(x, "name", getattr(x, "__name__", None)) for x in new_agent.tools])
break
if __name__ == "__main__":
asyncio.run(main())
Notes and how this addresses your issues
• Tools are always created with function_tool(name=..., description=...) so the LLM receives a clear schema.
• Tools are attached in a single batched await agent.update_tools(...) per agent to avoid intermediate inconsistent states.
• The transfer tool returns a new agent instance that already has tools attached (copied from the pre-built target agent), so when the model hands off to the new agent those tools exist and can be executed.
• Early prints inside tools and transfer construction help confirm whether the function is actually invoked.
Next steps
• Replace the placeholder session/LLM/STT/ TTS with your real components and run inside your environment.
• Add early-entry prints at top of each tool to confirm invocation in real sessions.
• If you still see hangs after this pattern, capture the model/tool negotiation logs (enable logging in code via logging.basicConfig(level=logging.DEBUG) if CLI flags are unavailable) and share the tool call payload and whether the server receives it.
If you want, I can adapt this minimal example to match your exact session setup (STT/LLM/TTS) — tell me which pieces you use and I’ll adjust.
Sources: Agents and handoffs | LiveKit docs | Tool definition and use | LiveKit docs