Tool Hooks

"""Tool hooks for the order-manager agent."""
from typing import Any
from connic import AbortTool

async def before(tool_name: str, params: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    """
    Called before every tool call.

    Args:
        tool_name: Name of the tool about to run (e.g. "get_order")
        params: Dict of parameters the LLM chose for the tool
        context: Shared run context dict (see Context docs)

    Returns:
        Parameter keys to add or replace. Return None to keep the original parameters.
    """
    # Block deletions for non-admin users
    if tool_name == "delete_order" and not context.get("is_admin"):
        raise AbortTool({"error": "Permission denied: only admins can delete orders"})

    return params

async def after(tool_name: str, params: dict[str, Any], result: Any, context: dict[str, Any]) -> Any:
    """
    Called after every tool call.

    Args:
        tool_name: Name of the tool that just ran
        params: The parameters the tool was called with
        result: The tool's return value
        context: Shared run context dict (see Context docs)

    Returns:
        Modified result (or original unchanged)
    """
    print(f"[hook] {tool_name}({params}) -> {result}")
    return result

"""Block specific tools based on runtime conditions."""
from typing import Any
from connic import AbortTool

async def before(tool_name: str, params: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    # Block destructive tools outside business hours
    from datetime import datetime
    hour = datetime.now().hour

    destructive = {"delete_order", "db_delete", "cancel_subscription"}
    if tool_name in destructive and not (9 <= hour < 17):
        raise AbortTool({
            "error": f"Tool '{tool_name}' is only available during business hours (9am-5pm)"
        })

    return params

"""The context parameter is optional."""
from typing import Any

# Without context - simpler signature
async def before(tool_name: str, params: dict[str, Any]) -> dict[str, Any]:
    return params

# With context - access run metadata and middleware values
async def after(tool_name: str, params: dict[str, Any], result: Any, context: dict[str, Any]) -> Any:
    run_id = context.get("run_id")
    return result

"""Sync hooks also work."""
from typing import Any

def before(tool_name: str, params: dict[str, Any]) -> dict[str, Any]:
    """Sync functions are automatically handled."""
    if "query" in params:
        params["query"] = params["query"].strip()
    return params

def after(tool_name: str, params: dict[str, Any], result: Any) -> Any:
    """Both sync and async are supported."""
    return result