https://livekit.io logo
Join Slack
Powered by
Hi, I want to implement a system in Agent wherein ...
# ask-ai
e
Hi, I want to implement a system in Agent wherein we can define required data for tool and MCP calls eg: Say I want to add a checkout tool (a POST webhook call to our servers), I want to define that the user must first submit their location, phone number and optionally their Name. If a tool is accidentally called without any data, there should be a way to transfer conttrol back to LLM without calling the webhook (webhook is still set to return a failure if this happens but I want conformation on agent side too). Also, same for MCP (if possible do not suggest converting MCPs into tool calls) What is the recommended way to do this? We already tried making a custom workflow using agent instruction update at runtime but that caused issues in that sometimes agent stops responding, context doesn't transfer or incporrect tools get called so not that.
t
Thanks for reaching out, Pratham. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.
Thanks — this is a very clear request. I’m responding to your question about enforcing required user data (location, phone number, optional name) before tool or MCP calls, and ensuring control returns to the LLM when data is missing without triggering your webhook. Short answer (recommended) • Enforce validation inside your function tools (the tool handler checks 
session.userdata
and/or its parameters). • If required fields are missing, raise a tool-level error (stop execution and return a clear error to the LLM). That prevents the webhook from being invoked and lets the LLM ask the user for the missing info. • For MCPs, do not convert critical actions to MCPs. Instead use function tools for critical webhook calls and keep MCPs for non-critical tasks. Use explicit agent instructions to discourage calling MCPs before data is collected. Why this approach works • Validation inside the tool is the single source of truth: the tool decides whether it’s safe to call the webhook. If validation fails, the tool never performs the POST. • Raising a tool error returns control to the LLM so it can prompt for missing data; this is more reliable than changing agent instructions at runtime (which you observed can break context or tool selection). • Storing collected values in session-level state (
session.userdata
) ensures subsequent tool calls can reuse the data reliably. Practical implementation outline 1. Collect data with dedicated tools ◦ Provide small tools the LLM can call to save 
location
, 
phone_number
, and 
name
into 
session.userdata
. ◦ These tools should acknowledge the value back to the user (so the LLM and user both know data is stored). 1. Validate in the checkout tool ◦ In the 
checkout
function tool, read 
context.session.userdata
and function parameters. ◦ Compute missing required fields. If any required field is missing, raise a tool-specific error (so the tool does not call the webhook). ◦ Only when validation passes, perform the webhook POST and return success/failure. 1. Clear, strict tool error messaging ◦ When raising the error, include which fields are missing and a short instruction (e.g., “Missing required data: location, phone_number — please ask user to provide these before retrying”). ◦ This enables the LLM to immediately ask the user for those fields and retry. 1. MCP guidance (do not convert critical MCPs) ◦ MCP calls are externally managed and harder to intercept. For critical actions (like checkout webhooks), keep those as function tools where you can validate before calling. ◦ For MCPs you must keep, add explicit agent instructions that require collecting required data first and mark the MCPs as non-critical or idempotent where possible. Example behavior (conceptual) • LLM tries to call `checkout`: ◦ Checkout tool reads 
session.userdata
, finds phone number missing → raises ToolError "Missing phone number". ◦ LLM receives the error, asks the user for the phone number (or calls 
provide_phone_number
). ◦ Once stored, the LLM calls 
checkout
again; tool validation passes and webhook runs. Why not runtime instruction updates • You already noticed problems: dynamic instruction updates can break the LLM’s context, cause it to stop responding, or cause incorrect tool calls. Validating at the tool-level avoids modifying agent instructions at runtime and is more deterministic. Advanced options (if you need stronger guards) • Implement a small intermediary service (a custom MCP or proxy) that will refuse webhook calls if required session metadata is missing — this is useful if multiple agents/services might call the same webhook and you want server-side enforcement. • Add unit/integration tests that simulate missing fields so you can validate agent behavior without relying on live users. Next steps / clarifying questions • Do your tools currently persist user data in a session object like 
session.userdata
(yes/no)? • Are your webhooks already implemented as function tools, or are some of them implemented as MCPs that you want to keep? • If you want, I can provide a concrete code snippet matching your agent SDK/language to validate and raise an error in the checkout tool. If you share which SDK/language and how you currently store session state, I’ll give a brief code example you can drop into your agent. Sources: Tool definition and use | LiveKit Documentation | Agent Builder | LiveKit Documentation | Testing and evaluation | LiveKit Documentation
e
Tools: 1. On frontend we have user submit which keys are required and which are optional in query, body etc. (Already present for query but being ignored on backend) 2. We will add a validation layer constructed based on inputs from frontend for which keys and what format those keys were required in from LLM # checkout_tool.py import requests from jsonschema import validate, ValidationError checkout_schema = { "type": "object", "properties": { "location": { "type": "string", "minLength": 5 }, "phoneNumber": { "type": "string", "pattern": "^[0-9+]{8,15}$" }, "name": { "type": "string", "minLength": 2 } }, "required": ["location", "phoneNumber"], "additionalProperties": False } def execute_checkout_tool(params: dict): """ Guarded execution of checkout tool. Prevents webhook call if validation fails. """ # 1️⃣ Validate tool arguments try: validate(instance=params, schema=checkout_schema) except ValidationError as e: return { "success": False, "transfer_to_llm": True, "error": "Validation failed", "details": e.message } # 2️⃣ Only execute webhook if valid try: response = requests.post( "https://your-server.com/webhook/checkout", json={ "location": params["location"], "phoneNumber": params["phoneNumber"], "name": params.get("name") }, timeout=5 ) response.raise_for_status() return { "success": True, "data": response.json() } except requests.RequestException as e: return { "success": False, "transfer_to_llm": True, "error": "Webhook failed", "details": str(e) } Note: Tool must not throw an error otherwise livekit will stop responding In system prompt we need to add something like: When a tool returns: - ok: false - error_type: missing_required_fields You must ask the user for the listed missing fields before retrying the tool. When a tool returns: - ok: false - error_type: webhook_failed Inform the user that checkout failed and ask if they would like to try again. For tools, like this right?
t
You're on the right track — your design is correct and follows robust patterns for preventing invalid tool calls and keeping the agent responsive. I’ll confirm why each piece is good, highlight two small fixes, and offer a couple of optional improvements you can adopt. What you’re doing right • Validating inputs with a JSON Schema before performing the webhook prevents invalid or malformed data from ever reaching your backend. • Returning a structured error object (instead of raising) keeps the agent loop alive and gives the LLM a clear signal to ask the user for missing/corrected fields. • Adding explicit system-prompt rules that map error responses to LLM behaviors (ask for missing fields, offer retry after webhook failure) closes the recovery loop and makes behavior predictable. • Distinguishing required vs optional fields in the schema and using explicit messages helps the LLM to ask for only what’s needed. Two small fixes / cautions 1. Phone number regex: your pattern string currently uses escaping that may be treated literally. Use a raw string or double-escape backslashes, e.g.: ◦ pattern: r"^[0-9+]{8,15}$" or in JSON schema: "pattern": "^[0-9+]{8,15}$" This avoids unintended pattern behavior. 1. Avoid ambiguous boolean fields: you used keys like 
"success": False
and 
"transfer_to_llm": True
. Keep a consistent field naming and semantics (see suggestion below) so the LLM and any orchestration logic can parse responses reliably. Suggested response format (more explicit and extensible) • success: boolean • error_type: one of ["validation_error", "webhook_failed", "webhook_timeout", null] • message: human readable summary • details: machine-readable details (e.g., list of missing fields or schema validation messages) Example:
Copy code
{
  "success": false,
  "error_type": "validation_error",
  "message": "Missing required fields: phoneNumber",
  "details": {
    "missing_fields": [
      "phoneNumber"
    ]
  }
}
This makes system-prompt rules and any automated routing easier to implement. Optional enhancements • Provide the JSON Schema to the LLM (via tool metadata or 
@function_tool
schema) so the model is more likely to construct valid calls up-front. • Log every tool call and returned validation failures for analytics — helps you refine the schema and prompt when the LLM repeatedly requests the same invalid values. • If you need stronger guarantees across services, add a lightweight server-side guard that rejects webhook requests lacking required session metadata. This is complementary to your client-side validation. • Normalize error types for finer-grained UX (e.g., 
webhook_timeout
vs 
webhook_4xx
), so the LLM can choose appropriate user messages (try again vs fix input). Edge cases to test • Partial inputs merged from session state + tool params (confirm merge logic and precedence). • Concurrent sessions or retries to ensure you don’t double-execute webhooks. • Unexpected additional properties (your schema already denies those — good). If you want, I can: • Convert your example into a ready-to-run function-tool stub (with proper regex handling and the normalized response format). • Draft the small system-prompt text that instructs the LLM to interpret your standardized error responses and the exact follow-up behavior you want. Would you like the function-tool stub or the exact system-prompt wording next? Sources: Tool definition and use | LiveKit Documentation | Agent Builder | LiveKit Documentation
e
How to add an interception for mcp calls, basically, a pythoin code should run before control goes to MCP?
t
You can run Python code before MCP calls by intercepting the agent’s tool-invocation flow. Two reliable options are: • Override the LLM output node (
llm_node
) and inspect generated tool calls; run your Python interception logic when you detect an MCP tool before allowing the tool call to proceed. • Wrap MCP tools with local function tools (a thin wrapper that runs your pre-checks then forwards to the real MCP), and expose only the wrapper to the LLM. Which to pick depends on scope: • Use llm_node override for a single centralized interception point that can handle every MCP tool. • Use wrapper tools when you want per-tool, fine-grained control or simpler logic per tool. Short examples and how they behave 1) Intercept via llm_node (centralized) • Where to hook: Implement 
llm_node
in your Agent subclass (this runs when the LLM generates content, including tool calls). • What it does: Inspect the LLM output stream; if a function/tool call is detected and the tool is an MCP tool, run your pre-MCP Python code (validation, logging, transforms). Then allow the stream to continue so the MCP call proceeds. • Good for: Global policies (validation, rate-limiting, audit logs, gating), and cases where you must inspect LLM intent before any tool is invoked. Conceptual sketch:
Copy code
async def llm_node(self, chat_ctx, tools, model_settings=None):
    async for chunk in Agent.default.llm_node(self, chat_ctx, tools, model_settings):
        if getattr(chunk, "function_call", None):
            tool_name = chunk.function_call.name
            if tool_name in self.mcp_tool_names:
                # Run your pre-MCP Python logic here
                await self.pre_mcp_hook(tool_name, chunk.function_call.arguments)
        yield chunk
2) Wrap MCP tools (per-tool wrappers) • Where to hook: Register a local 
@function_tool
wrapper in place of (or in front of) the MCP tool the LLM would call. • What it does: Run your Python checks/validation inside the wrapper. If checks pass, call the MCP tool (via the MCP client or forward request); if checks fail, return a structured error to the LLM (do not raise). • Good for: When you only need interception for specific tools or want to encapsulate different logic per tool. Conceptual sketch:
Copy code
@function_tool()
async def checkout_wrapper(context, params):
    # run pre-checks
    if not valid(params):
        return {"success": False, "error_type": "validation_error", "details": ...}
    # forward to MCP tool / call MCP endpoint
    result = await call_mcp_tool("checkout", params)
    return result
Practical considerations • MCP tools are normally discovered from configured MCP servers. If you wrap them, expose only the wrapper tool to the agent (so the raw MCP tool isn’t called directly). • Keep returned errors structured (e.g., error_type, message, details). The LLM can then ask for missing fields or prompt retry without the agent stalling. • Do not raise unhandled exceptions in tool code — return structured failures so the agent and LLM remain responsive. • If you need server-side guarantees, also add server-side guards (the wrapper or backend MCP service) to prevent accidental direct calls. Questions to give you a concrete example • Do you want to intercept all MCP calls globally, or only certain tools (e.g., checkout)? • Which SDK / language and Agent class are you using (Python agent subclass, exact version)? I can give a full runnable example for your setup. • Are the MCP tools already registered via 
mcp_servers
in your 
AgentSession
, or do you control the MCP server endpoints directly? Tell me which approach you prefer (global 
llm_node
interception vs per-tool wrapper), your SDK/language, and I’ll provide a concrete code sample you can drop into your agent. Sources: Tool definition and use | LiveKit Documentation | Pipeline nodes and hooks | LiveKit Documentation | External data and RAG | LiveKit Documentation | MCP Agent | LiveKit Documentation | LLM Output Replacement | LiveKit Documentation
e
async def handle_mcp_call(tool_name: str, arguments: dict): required = MCP_REQUIRED_FIELDS.get(tool_name, []) missing = [field for field in required if not arguments.get(field)] if missing: # Do NOT call MCP return { "role": "system", "content": ( f"The action '{tool_name}' cannot be executed. " f"Missing required fields: {', '.join(missing)}. " f"Ask the user to provide them." ) } # If valid → allow normal MCP execution return await session.call_mcp(tool_name, arguments) Don't you allowcustom functions, something like what I shared
t
Yes — you can run custom Python validation exactly like the snippet you shared. You should implement that logic as a wrapper function/tool inside your agent so the LLM never gets direct access to the raw MCP call. That wrapper will validate inputs, return a structured failure to the LLM when required fields are missing, and only forward to the MCP server when validation passes. Key points to implement safely and reliably • Expose only your wrapper function/tool to the LLM, not the raw MCP tool. This ensures the LLM cannot invoke MCPs directly. • Perform validation inside the wrapper and return a structured response (do not raise unhandled exceptions). Structured responses let the LLM ask for missing fields or retry without the agent stalling. • If validation passes, forward to your MCP server via an HTTP/SDK call from the wrapper. There is no built-in single 
session.call_mcp()
method you must rely on; call the MCP endpoint from your code. • Keep error types consistent and machine-friendly (e.g., error_type: "validation_error" / "webhook_failed"), so your system prompt can teach the LLM how to respond. Concrete pattern (conceptual) 1. Define required fields map: MCP_REQUIRED_FIELDS = {"checkout": ["location","phoneNumber"], ...} 1. Wrapper tool: ◦ Validate arguments using your schema or simple presence checks. ◦ If missing → return: { "success": False, "error_type": "validation_error", "message": "Missing required fields: phoneNumber", "details": {"missing_fields": ["phoneNumber"]} } • If valid → call MCP via HTTP/SDK and return MCP result or structured webhook failure: { "success": True, "data": {...} } or { "success": False, "error_type": "webhook_failed", "message": "Remote error: timeout" } 1. System prompt guidance: ◦ Instruct the LLM to request missing fields when it receives a validation_error. ◦ Instruct it to inform the user and offer retry on webhook_failed. Why this is preferred • Direct interception hooks into the LLM dispatch are not necessary; the wrapper is the supported, deterministic pattern. • Wrappers let you run any synchronous or async Python logic (validation, enrichment, logging, rate-limiting, additional checks) before an MCP call. • It prevents accidental direct MCP invocation that you cannot validate or control. If you want, I can: • Convert your snippet into a complete async wrapper using aiohttp/aiohttp.ClientSession that calls your MCP endpoint. • Provide the exact system-prompt text that teaches the LLM how to react to your structured error responses. • Show a pattern to programmatically generate these wrappers if you have many MCP tools. Which would you like next: the full async wrapper implementation, or the system-prompt wording? Sources: Tool definition and use | LiveKit Documentation
e
handle_mcp_call is this an actual function that I can override? What I shared was an example code
t
Good question — thank you for clarifying. Short answer • No: 
handle_mcp_call
is not a built-in overridable method in LiveKit Agents. Your snippet is a perfectly valid user-defined pattern, but it’s not a framework hook you can override. • Yes: you can implement the same behavior by either (A) exposing wrapper function-tools that perform validation and then call the MCP endpoint, or (B) intercepting LLM output by overriding 
llm_node
(or other lifecycle hooks) and running your validation before allowing the MCP call to proceed. What I recommend (clear, reliable, and easy to maintain) • Implement wrapper function-tools in your agent that the LLM calls instead of the raw MCP tools. In those wrappers you: ◦ Validate inputs (your JSON Schema / required-fields check). ◦ If validation fails, return a structured error to the LLM (do not raise). ◦ If validation passes, call the MCP server (HTTP/SDK) and return the MCP result or a structured failure. • This pattern gives you deterministic control, simple error handling, and no need to rely on internal hooks. Why wrapper tools are preferable • They prevent the LLM from calling the raw MCP tools directly. • They let you run arbitrary Python (validation, enrichment, logging, rate-limits) synchronously or asynchronously. • They return structured responses the LLM can act on (ask for missing fields, retry, etc.). • They avoid modifying LLM internals or runtime instruction updates that can cause instability. When to use 
llm_node
override instead • Use 
llm_node
if you need a global interception point (e.g., one place to apply policy for every tool call, dynamic routing, or centralized audit), and you are comfortable inspecting LLM-generated tool calls in streaming output. • 
llm_node
approach is more advanced and requires careful handling of streaming chunks and function-call detection. Concrete options and brief examples 1) Wrapper function-tool (recommended) • Expose 
checkout_wrapper
as a function-tool the LLM can call. • Inside it run your required-field logic (your example code). • If valid, perform an async HTTP call to your MCP endpoint and return the response. 2) llm_node interception (central control) • Override 
llm_node
in your Agent subclass. • Inspect the LLM’s generated function call chunk; when it targets an MCP tool, run your pre-check hook. • If checks fail, inject a system message or return a structured reply so LLM asks for missing info instead of proceeding. Notes / gotchas • There is no built-in 
session.call_mcp()
method you can call directly — you’ll call your MCP server via HTTP or the MCP client SDK from your wrapper. • Never raise unhandled exceptions in tool code; always return a structured failure so the agent and LLM remain responsive. • Keep your error format consistent (e.g., error_type / message / details) so the LLM follows your system-prompt rules. If you want, I can: • Convert your example into a full async wrapper tool implementation (aiohttp + JSON Schema validation) you can drop into your agent. • Or provide a 
llm_node
override example that detects MCP function calls and runs your pre-check hook. Which would you prefer — the wrapper implementation, or an 
llm_node
interception example?
e
Copy code
async def llm_node(self, chat_ctx, tools, model_settings=None):
    async for chunk in Agent.default.llm_node(self, chat_ctx, tools, model_settings):
        if getattr(chunk, "function_call", None):
            tool_name = chunk.function_call.name
            if tool_name in self.mcp_tool_names:
                # Run your pre-MCP Python logic here
                await self.pre_mcp_hook(tool_name, chunk.function_call.arguments)
        yield chunk
This right? we are talking about mcps added in mcp_server, not as a function tool
t
Yes — that snippet is the right idea for intercepting MCP tools that were loaded via 
mcp_servers
(not exposed as local function tools). Overriding 
llm_node
lets you inspect the LLM’s generated output (including function/tool calls) and run Python logic before the tool invocation proceeds. A few important details and caveats to make it reliable: • It intercepts generated function calls, not the raw HTTP call to the MCP server. The LLM outputs a function-call intent; your agent runtime will then dispatch that intent to the MCP client. By checking chunks in 
llm_node
you can run pre-checks and take action before dispatch. • You must detect the function-call shape used by your LLM streaming chunks. In many tool-call streaming implementations the chunk contains a 
function_call
object with 
.name
and 
.arguments
. Your snippet checks that correctly with 
getattr(chunk, "function_call", None)
. • Keep 
llm_node
async and streaming-safe: do not block long-running synchronous work inside it. Use awaitable async hooks (e.g., 
await self.pre_mcp_hook(...)
) and short timeouts for external calls. • If your pre-check wants to stop the MCP call, you need to convert that into an action the agent/LLM can respond to. You can: ◦ Inject a system/tool message back into the conversation explaining validation failure (so LLM asks the user for missing info), or ◦ Modify the stream to yield a replacement chunk (e.g., a system message) instead of yielding the original function_call chunk. How exactly to inject/replace depends on your agent runtime’s streaming interfaces; often returning a system message or skipping yielding the function_call chunk prevents dispatch. • Logging/observability: log intercepted attempts and missing-field details so you can refine prompts and schema. • Security/consistency: ensure detection of MCP tools (your 
self.mcp_tool_names
) is accurate and maintained when MCP servers change tools. Example expanded pattern (conceptual)
Copy code
async def llm_node(self, chat_ctx, tools, model_settings=None):
    async for chunk in Agent.default.llm_node(self, chat_ctx, tools, model_settings):
        fc = getattr(chunk, "function_call", None)
        if fc:
            tool_name = fc.name
            if tool_name in self.mcp_tool_names:
                ok, reason = await self.pre_mcp_hook(tool_name, fc.arguments)
                if not ok:
                    # Instead of allowing the MCP call, return a system message
                    # so the LLM asks the user for required inputs.
                    yield SystemMessage(
                        content=f"The action '{tool_name}' cannot run: {reason}. Please ask the user for the missing fields."
                    )
                    continue  # do not yield the original function_call chunk
        yield chunk
If you want, I can: • Give a concrete runnable example tailored to the exact agent SDK you’re using (showing how to inject a system message or how to abort the function_call dispatch). • Show a safe 
pre_mcp_hook
implementation that performs JSON Schema validation and returns a normalized result format for the LLM. Which would you prefer — a runnable example for aborting the MCP call and injecting a system message, or a 
pre_mcp_hook
implementation that validates against schemas?
Open in Slack
PreviousNext
Powered by