What it is

When the Anthropic Messages API returns a tool_use block in its response, tool_executor.execute is the only function allowed to turn that into an actual side effect. It holds the registry, applies the safety rules, and formats the result back into the tool_result shape Anthropic expects.

Registered tools

Tool	Module	Purpose
`venue_search`	`tools/venue_search.py`	Intent-based venue search across 35 patterns.
`create_booking`	`tools/booking.py`	Create a booking. Routes through ClawPulse when the venue is on the network.
`check_booking`	`tools/booking.py`	Look up the current status of a booking the user already made.
`cancel_booking`	`tools/booking.py`	Cancel an existing booking and fire the associated state transition.
`get_user_bookings`	`tools/booking.py`	List the user's bookings past and upcoming.
`rate_booking`	`tools/booking.py`	Submit a post-visit rating with comment.
`set_reminder`	`tools/reminders.py`	Schedule a reminder delivered by the deliver_reminders scheduler job.
`save_preference`	`tools/crm_write.py`	Persist a taste / dietary / lifestyle preference to UserMemory.
`dupe_search_by_text`	`tools/dupe_search.py`	Find cheaper alternatives across Shopee, Tokopedia, Lazada.
`set_price_watcher`	`tools/dupe_search.py`	Watch a product and alert on price drops.
`calorie_scan_by_text`	`tools/calorie_scan.py`	Resolve a text meal description to macros via Nutritionix.
`get_nutrition_summary`	`tools/calorie_scan.py`	Daily calorie + macro summary for the user's food diary.
`get_ride`	`tools/transport.py`	Uber / Grab / GoJek deep links for a route.
`get_delivery_quote`	`tools/transport.py`	Lalamove delivery quote (read-only).
`book_lalamove_delivery`	`tools/transport.py`	Confirm a Lalamove delivery. Requires explicit user confirmation.
`generate_card`	`tools/generate_card.py`	Generate a shareable interaction card with a public URL.

A 17th optional tool, knowledge_lookup, is registered only when KNOWLEDGE_LOOKUP_URL is set.

Registration

Tools are registered at import time. The registry is a module-level dict keyed by tool name. Optional tools conditionally append their schema to TOOL_SCHEMAS so the LLM only ever sees the tools that are actually callable.

pythonapp/core/tool_executor.py — registration

from app.core.tools import TOOL_REGISTRY
from app.core.tools import venue_search as _vs
from app.core.tools import booking as _bk
from app.core.tools import reminders as _rm
from app.core.tools import crm_write as _crm
from app.core.tools import dupe_search as _ds
from app.core.tools import calorie_scan as _cal
from app.core.tools import transport as _tr
from app.core.tools import generate_card as _gc

# Register tool functions
TOOL_REGISTRY["venue_search"] = _vs.venue_search
TOOL_REGISTRY["create_booking"] = _bk.create_booking
TOOL_REGISTRY["check_booking"] = _bk.check_booking
# ...

# Optional public-knowledge tool — only registered when configured.
if _settings.KNOWLEDGE_LOOKUP_URL:
    TOOL_REGISTRY["knowledge_lookup"] = _kl.knowledge_lookup

The execute function

The execution path is intentionally boring. Safe input gets prepared, a 5-second timeout is applied, exceptions are caught and returned as strings, and timeouts turn into a specific error message.

pythonapp/core/tool_executor.py — execute

async def execute(tool_call: ToolCall, user_id: uuid.UUID, db: AsyncSession) -> str:
    fn = TOOL_REGISTRY.get(tool_call.name)
    if not fn:
        return f"Error: Unknown tool '{tool_call.name}'"

    try:
        # Strip reserved keys from LLM-generated input to prevent injection
        safe_input = {k: v for k, v in tool_call.input.items() if k not in ("user_id", "db")}

        # Validate tool parameters: constrain string lengths to prevent DoS
        _MAX_PARAM_LENGTH = 1000
        for k, v in safe_input.items():
            if isinstance(v, str) and len(v) > _MAX_PARAM_LENGTH:
                safe_input[k] = v[:_MAX_PARAM_LENGTH]
            elif isinstance(v, list) and len(v) > 50:
                safe_input[k] = v[:50]

        result = await asyncio.wait_for(
            fn(**safe_input, user_id=str(user_id), db=db),
            timeout=5.0,
        )
        return result
    except asyncio.TimeoutError:
        return f"Error: Tool '{tool_call.name}' timed out after 5 seconds"
    except Exception as e:
        return f"Error executing {tool_call.name}: {e}"

Safety rules

Reserved keys are stripped — user_id and dbare passed by the executor, not by the LLM. If the LLM tried to inject them, they're silently discarded.
Strings clamp to 1000 chars to prevent token-based DoS on downstream APIs.
Lists clamp to 50 items for the same reason.
5 second timeout via asyncio.wait_for. Tools that need longer must be split into smaller steps or moved to a background job.
Exceptions are swallowed into a readable error string instead of bubbling back to the LLM. This keeps the conversation alive and lets the model recover.

Running a batch

The LLM gateway calls run(tool_calls, user_id, db)which just sequentially executes each tool and shapes the results into Anthropic's tool_result content blocks.

pythonapp/core/tool_executor.py — run

async def run(tool_calls: list[ToolCall], user_id: uuid.UUID, db: AsyncSession) -> list[dict]:
    """Execute all tool calls and return Anthropic-format tool_result content blocks."""
    results = []
    for tc in tool_calls:
        output = await execute(tc, user_id, db)
        results.append({
            "type": "tool_result",
            "tool_use_id": tc.id,
            "content": output,
        })
    return results

Adding a new tool

Write the function in app/core/tools/your_tool.py. Signature: async def your_tool(**kwargs, user_id: str, db: AsyncSession) -> str
Add its schema to TOOL_SCHEMAS in app/core/tools/__init__.py.
Register the function in tool_executor.py under a unique name.
Write a test in tests/ that covers the happy path and at least one failure path.

Tool executor.