Claude Tool Use

import Anthropic from "@anthropic-ai/sdk";

// Define the tool schemas Claude will use
const tools: Anthropic.Tool[] = [
  {
    name: "process_refund",
    description: "Process a customer refund. Requires human approval.",
    input_schema: {
      type: "object" as const,
      properties: {
        orderId: { type: "string", description: "The order ID to refund" },
        amount: { type: "number", description: "Refund amount in USD" },
        reason: { type: "string", description: "Reason for the refund" },
      },
      required: ["orderId", "amount"],
    },
  },
  {
    name: "send_email",
    description: "Send an email to a customer.",
    input_schema: {
      type: "object" as const,
      properties: {
        to: { type: "string" },
        subject: { type: "string" },
        body: { type: "string" },
      },
      required: ["to", "subject", "body"],
    },
  },
];

import Anthropic from "@anthropic-ai/sdk";
import { CheqpointClient, RejectedError } from "@cheqpoint/sdk";

const anthropic = new Anthropic();
const cheqpoint = new CheqpointClient({ apiKey: process.env.CHEQPOINT_CONNECTION_KEY! });

// Tools that always require human review
const SENSITIVE_TOOLS = new Set(["process_refund", "transfer_funds", "delete_record", "send_email"]);

async function runAssistant(userMessage: string) {
  const messages: Anthropic.MessageParam[] = [{ role: "user", content: userMessage }];

  while (true) {
    const response = await anthropic.messages.create({
      model: "claude-opus-4-6",
      max_tokens: 4096,
      tools,
      messages,
    });

    if (response.stop_reason === "end_turn") {
      const textBlock = response.content.find((b) => b.type === "text");
      return textBlock?.type === "text" ? textBlock.text : "";
    }

    const toolResults: Anthropic.ToolResultBlockParam[] = [];

    for (const block of response.content) {
      if (block.type !== "tool_use") continue;

      let result: unknown;

      if (SENSITIVE_TOOLS.has(block.name)) {
        try {
          const approval = await cheqpoint.checkpoint({
            action: block.name,
            summary: `Claude wants to call ${block.name}`,
            details: block.input as Record<string, unknown>,
            riskScore: 0.8,
          });
          // checkpoint() throws RejectedError if declined — reaching here means approved
          const payload = approval.modifiedDetails ?? (block.input as Record<string, unknown>);
          result = await executeTool(block.name, payload);
        } catch (e) {
          if (e instanceof RejectedError) {
            result = { error: `Not approved: ${e.message}` };
          } else {
            result = { error: e instanceof Error ? e.message : "Unknown error" };
          }
        }
      } else {
        result = await executeTool(block.name, block.input as Record<string, unknown>);
      }

      toolResults.push({
        type: "tool_result",
        tool_use_id: block.id,
        content: JSON.stringify(result),
      });
    }

    messages.push({ role: "assistant", content: response.content });
    messages.push({ role: "user", content: toolResults });
  }
}

import os
import anthropic
from cheqpoint import CheqpointClient, RejectedError

client = anthropic.Anthropic()
cheq = CheqpointClient(api_key=os.environ["CHEQPOINT_CONNECTION_KEY"])

SENSITIVE_TOOLS = {"process_refund", "transfer_funds", "delete_record"}

tools = [
    {
        "name": "process_refund",
        "description": "Process a customer refund. Requires human approval.",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string"},
                "amount": {"type": "number"},
            },
            "required": ["order_id", "amount"],
        },
    }
]

def run_assistant(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            return next((b.text for b in response.content if b.type == "text"), "")

        tool_results = []
        for block in response.content:
            if block.type != "tool_use":
                continue

            if block.name in SENSITIVE_TOOLS:
                try:
                    result_obj = cheq.checkpoint(
                        action=block.name,
                        summary=f"Claude wants to {block.name}",
                        details=block.input,
                        risk_score=0.8,
                    )
                    payload = result_obj.modified_details or block.input
                    output = execute_tool(block.name, payload)
                except RejectedError as e:
                    output = {"error": f"Rejected: {e}"}
            else:
                output = execute_tool(block.name, block.input)

            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": str(output),
            })

        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})