Tool Call Validation and Approval Guide

import json
from os import environ

from anthale import Anthale

client = Anthale(api_key=environ["ANTHALE_API_KEY"])
TOOL_OUTPUT_POLICY_ID = environ["ANTHALE_TOOL_OUTPUT_POLICY_ID"]

ALLOWED_TOOLS = {"search_docs", "create_ticket"}
ALLOWED_PRIORITIES = {"low", "medium", "high"}

def execute_tool(name: str, arguments: dict) -> dict:
raise NotImplementedError("Connect this helper to your real tool runner.")

def validate_tool_call(proposed_call: dict, approved: bool) -> dict:
if proposed_call["name"] not in ALLOWED_TOOLS:
raise RuntimeError("Tool is not allowed for this workflow.")

    if proposed_call["name"] == "create_ticket":
        priority = proposed_call["arguments"]["priority"]
        if priority not in ALLOWED_PRIORITIES:
            raise RuntimeError("Priority must be low, medium, or high.")
        if not approved:
            raise RuntimeError("create_ticket requires explicit approval.")

    return proposed_call

def guard_tool_output(tool_name: str, tool_output: dict) -> dict:
response = client.organizations.policies.enforce(
policy_identifier=TOOL_OUTPUT_POLICY_ID,
direction="input",
messages=[{"role": "user", "content": json.dumps(tool_output)}],
metadata={
"boundary": "tool_output",
"feature": "support-agent",
"toolName": tool_name,
"sourceType": "tool_output",
},
)
if response.action == "block":
raise RuntimeError(f"Anthale blocked output from {tool_name}.")

    safe_message = response.redacted_messages[0] if response.action == "redact" else {"content": json.dumps(tool_output)}
    return json.loads(safe_message["content"])

def handle_tool_proposal(proposed_call: dict, approved: bool) -> dict:
validated_call = validate_tool_call(proposed_call, approved)
raw_tool_output = execute_tool(validated_call["name"], validated_call["arguments"])
return guard_tool_output(validated_call["name"], raw_tool_output)

import Anthale from "anthale"

const client = new Anthale({ apiKey: process.env.ANTHALE_API_KEY })
const TOOL_OUTPUT_POLICY_ID = process.env.ANTHALE_TOOL_OUTPUT_POLICY_ID!

const ALLOWED_TOOLS = new Set(["search_docs", "create_ticket"])
const ALLOWED_PRIORITIES = new Set(["low", "medium", "high"])

type ToolCall = {
  name: string
  arguments: Record<string, string>
}

async function executeTool(name: string, arguments_: Record<string, string>): Promise<Record<string, unknown>> {
  throw new Error("Connect this helper to your real tool runner.")
}

function validateToolCall(proposedCall: ToolCall, approved: boolean): ToolCall {
  if (!ALLOWED_TOOLS.has(proposedCall.name)) {
    throw new Error("Tool is not allowed for this workflow.")
  }

  if (proposedCall.name === "create_ticket") {
    if (!ALLOWED_PRIORITIES.has(proposedCall.arguments.priority)) {
      throw new Error("Priority must be low, medium, or high.")
    }
    if (!approved) {
      throw new Error("create_ticket requires explicit approval.")
    }
  }

  return proposedCall
}

async function guardToolOutput(toolName: string, toolOutput: Record<string, unknown>): Promise<Record<string, unknown>> {
  const response = await client.organizations.policies.enforce(TOOL_OUTPUT_POLICY_ID, {
    direction: "input",
    messages: [{ role: "user", content: JSON.stringify(toolOutput) }],
    metadata: {
      boundary: "tool_output",
      feature: "support-agent",
      toolName,
      sourceType: "tool_output",
    },
  })

  if (response.action === "block") {
    throw new Error(`Anthale blocked output from ${toolName}.`)
  }

  const safeContent =
    response.action === "redact" ? response.redacted_messages[0].content : JSON.stringify(toolOutput)

  return JSON.parse(safeContent) as Record<string, unknown>
}

export async function handleToolProposal(proposedCall: ToolCall, approved: boolean): Promise<Record<string, unknown>> {
  const validatedCall = validateToolCall(proposedCall, approved)
  const rawToolOutput = await executeTool(validatedCall.name, validatedCall.arguments)
  return guardToolOutput(validatedCall.name, rawToolOutput)
}