how can i get tool name invoked and post to webhook ?

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

dynamically

Thanks — I hear you want a dynamic, non‑hardcoded way to capture whichever tool the agent invokes and post that info to a webhook. I can provide a ready solution, but first a couple quick clarifying questions so I give code that fits your setup: 1. Do you want the webhook POST to originate from the server-side agent process or from the client/browser that runs a client tool? 2. Which SDK/language are you using for your agent code: Python or Node.js? 3. Do you want to include the full tool arguments and result in the webhook, or only the tool name (and maybe an ID/timestamp)? 4. Do you need authentication on the webhook (e.g., add an Authorization header or HMAC signature)? If you prefer, tell me “server / Python / include args+result / HMAC” (or similar) and I’ll return a complete, dynamic example that: • Detects the invoked tool name automatically (no hardcoding), • Builds a payload with args/result/timestamp, • Sends the webhook with optional auth and retry handling.

server-side agent

Great — thanks for confirming. Below is a server-side Python pattern that dynamically captures whichever

function_tool

is invoked, includes the full tool arguments and result, and posts that payload to your webhook. It adds optional HMAC signing, retry logic, and non-blocking behavior so webhook failures won’t crash the tool unless you want them to. If you’d like any adjustments (no HMAC, synchronous blocking post, different retry policy), tell me and I’ll adapt it. 1) Core idea • Use a decorator wrapper around your tool functions to intercept the call, capture: tool name, args, kwargs, result (or exception), timestamp, and an execution id. • Post that payload to your webhook asynchronously with retries and optional HMAC auth. • Return the original result (or re-raise) so agent logic is unchanged. 2) Example implementation

import asyncio
import aiohttp
import functools
import json
import time
import uuid
import hmac
import hashlib
from typing import Any, Callable

WEBHOOK_URL = "<https://your-webhook.url/path>"
WEBHOOK_SECRET = b"your_secret_here"  # set to None to disable HMAC
MAX_RETRIES = 3
RETRY_BACKOFF = 1.5  # multiplier

def sign_payload(secret: bytes, payload_bytes: bytes) -> str:
    mac = hmac.new(secret, payload_bytes, digestmod=hashlib.sha256)
    return mac.hexdigest()

async def post_with_retries(url: str, json_payload: dict, headers: dict = None):
    headers = headers or {}
    attempt = 0
    delay = 1.0
    async with aiohttp.ClientSession() as session:
        while attempt < MAX_RETRIES:
            try:
                async with <http://session.post|session.post>(url, json=json_payload, headers=headers, timeout=10) as resp:
                    # consider 2xx as success
                    if 200 <= resp.status < 300:
                        return await resp.text()
                    # treat non-2xx as retryable (customize if needed)
                    text = await resp.text()
                    attempt += 1
                    await asyncio.sleep(delay)
                    delay *= RETRY_BACKOFF
            except asyncio.CancelledError:
                raise
            except Exception:
                attempt += 1
                await asyncio.sleep(delay)
                delay *= RETRY_BACKOFF
    # final failure; return None or raise depending on how you want to handle it
    return None

def webhook_notify_decorator(send_to_webhook: bool = True, fail_on_webhook_error: bool = False):
    """
    Decorator to wrap function tools. Captures tool name, args, kwargs, result/exception,
    and posts to webhook. If fail_on_webhook_error is True, exceptions in posting will be raised.
    """
    def decorator(fn: Callable):
        @functools.wraps(fn)
        async def wrapper(*args, **kwargs):
            invocation_id = str(uuid.uuid4())
            tool_name = fn.__name__
            timestamp = time.time()

            payload = {
                "invocation_id": invocation_id,
                "tool": tool_name,
                "args": args,        # may need serialization
                "kwargs": kwargs,    # may need serialization
                "timestamp": timestamp,
            }

            try:
                # execute the original tool
                result = await fn(*args, **kwargs)
                payload["result"] = result
                payload["status"] = "success"
            except Exception as exc:
                payload["result"] = {"error": str(exc)}
                payload["status"] = "error"
                # post the webhook about the failure, then re-raise
                if send_to_webhook:
                    payload_bytes = json.dumps(payload, default=str).encode("utf-8")
                    headers = {"Content-Type": "application/json"}
                    if WEBHOOK_SECRET:
                        headers["X-Signature"] = sign_payload(WEBHOOK_SECRET, payload_bytes)
                    # run post async and optionally wait/raise
                    coro = post_with_retries(WEBHOOK_URL, json.loads(payload_bytes), headers=headers)
                    task = asyncio.create_task(coro)
                    if fail_on_webhook_error:
                        resp = await task
                        if resp is None:
                            raise RuntimeError("Webhook POST failed after retries")
                raise

            # success: send webhook in background unless configured to wait
            if send_to_webhook:
                # serialize args/kwargs/result safely
                try:
                    payload_serializable = {
                        "invocation_id": invocation_id,
                        "tool": tool_name,
                        "args": json.loads(json.dumps(payload["args"], default=str)),
                        "kwargs": json.loads(json.dumps(payload["kwargs"], default=str)),
                        "result": json.loads(json.dumps(payload["result"], default=str)),
                        "timestamp": timestamp,
                        "status": payload["status"],
                    }
                except Exception:
                    # fallback: stringify everything
                    payload_serializable = {
                        "invocation_id": invocation_id,
                        "tool": tool_name,
                        "args": str(payload["args"]),
                        "kwargs": str(payload["kwargs"]),
                        "result": str(payload["result"]),
                        "timestamp": timestamp,
                        "status": payload["status"],
                    }

                payload_bytes = json.dumps(payload_serializable).encode("utf-8")
                headers = {"Content-Type": "application/json"}
                if WEBHOOK_SECRET:
                    headers["X-Signature"] = sign_payload(WEBHOOK_SECRET, payload_bytes)

                # post in background; do not block agent response
                task = asyncio.create_task(post_with_retries(WEBHOOK_URL, payload_serializable, headers=headers))
                # optionally await task here if you want to fail tool on webhook error:
                # resp = await task
            return result
        return wrapper
    return decorator

3) How to use it with your LiveKit function tools • Wrap your function tools with both the agent

@function_tool()

and the webhook decorator. Order matters: decorate with the webhook wrapper outermost so the wrapper observes the actual invocation and result. Example:

from livekit.agents import function_tool, Agent

class MyAgent(Agent):

    @webhook_notify_decorator(send_to_webhook=True, fail_on_webhook_error=False)
    @function_tool()
    async def lookup_weather(self, location: str) -> dict:
        # realistic implementation...
        result = {"weather": "sunny", "temperature_f": 70}
        return result

    @webhook_notify_decorator(send_to_webhook=True)
    @function_tool()
    async def compute_sum(self, a: int, b: int) -> int:
        return a + b

4) Notes and recommendations • Serialization: Arguments or results that include complex objects may not be directly JSON serializable. The example uses json.dumps(..., default=str) as a safe fallback. If you have structured types, add custom serializers. • Security: The example shows adding an

X-Signature

HMAC header. Verify this on the webhook receiver side with the same secret. • Reliability: Background posting (create_task) prevents slowing agent responses. If webhook delivery is critical, set

fail_on_webhook_error=True

and await the post task (or implement durable queuing). • Logging: Log webhook failures and responses for debugging. Consider exposing metrics on success/failure rates. • Concurrency: aiohttp ClientSession is created per call in this example; for high throughput, consider a long-lived session shared across calls. If you want, I can: • Provide a version that awaits the webhook post and raises on failure (blocking behavior). • Show a Node.js equivalent. • Add an example that includes the agent/room/job context (e.g., add room id / participant id) into the payload if you store that in the agent context. Which of those would help next?