Claude API Tool Use Practical Guide

Problem

Required Tools

Solution Steps

# ============================================
# Python SDK installation and initialization
# ============================================
pip install anthropic

# Set API key as environment variable (never hardcode in source!)
# Linux/Mac:
export ANTHROPIC_API_KEY="sk-ant-api03-..."

# Windows PowerShell:
$env:ANTHROPIC_API_KEY="sk-ant-api03-..."

# Python client initialization and test
import anthropic

client = anthropic.Anthropic()  # Automatically uses ANTHROPIC_API_KEY env var

# Connection test
message = client.messages.create(
    model="claude-sonnet-4-5-20250514",
    max_tokens=100,
    messages=[{"role": "user", "content": "Hello!"}]
)
print(message.content[0].text)  # If you see a response, setup is complete

# ============================================
# TypeScript SDK installation and initialization
# ============================================
# npm install @anthropic-ai/sdk
#
# import Anthropic from "@anthropic-ai/sdk";
# const client = new Anthropic();  // Automatically uses ANTHROPIC_API_KEY env var

import anthropic

client = anthropic.Anthropic()

# ============================================
# Tool definitions - parameter specs via JSON Schema
# ============================================
tools = [
    {
        "name": "get_weather",
        "description": (
            "Retrieves current weather information (temperature, humidity, conditions) "
            "for a specified city. City names can be provided in English or local language "
            "(e.g., 'Seoul', 'Tokyo', 'New York'). "
            "Use this tool when the user asks about weather, temperature, or climate."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "City name to check weather for (e.g., 'Seoul', 'Tokyo', 'New York')"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature unit. Defaults to celsius"
                }
            },
            "required": ["city"]
        }
    },
    {
        "name": "manage_schedule",
        "description": (
            "View (list), add, or delete user schedules. "
            "Supports checking schedules for a specific date, registering new events, "
            "and removing existing events. "
            "Use this tool when the user mentions schedules, appointments, meetings, or events."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "action": {
                    "type": "string",
                    "enum": ["list", "add", "delete"],
                    "description": "Action to perform: list (view), add (create), delete (remove)"
                },
                "date": {
                    "type": "string",
                    "description": "Target date (YYYY-MM-DD format). e.g., '2025-05-15'"
                },
                "title": {
                    "type": "string",
                    "description": "Event title (required when action is 'add')"
                },
                "time": {
                    "type": "string",
                    "description": "Event time (HH:MM format, optional when action is 'add')"
                },
                "schedule_id": {
                    "type": "string",
                    "description": "ID of the schedule to delete (required when action is 'delete')"
                }
            },
            "required": ["action", "date"]
        }
    }
]

# ============================================
# API call test with tools included
# ============================================
response = client.messages.create(
    model="claude-sonnet-4-5-20250514",
    max_tokens=1024,
    tools=tools,
    messages=[
        {"role": "user", "content": "What's the weather like in Seoul?"}
    ]
)

# Analyze the response
print(f"stop_reason: {response.stop_reason}")
# -> "tool_use" (Claude is requesting a tool call)

for block in response.content:
    if hasattr(block, "text"):
        print(f"Text: {block.text}")
    if block.type == "tool_use":
        print(f"Tool call: {block.name}")
        print(f"Parameters: {block.input}")
        print(f"tool_use_id: {block.id}")
# -> Tool call: get_weather
# -> Parameters: {"city": "Seoul"}
# -> tool_use_id: "toolu_01Abc..."

import anthropic
import json

client = anthropic.Anthropic()

# ============================================
# 1. Define actual tool execution functions
# ============================================
def get_weather(city: str, unit: str = "celsius") -> dict:
    """In practice, this would call a weather API like OpenWeatherMap."""
    # Example data (replace with actual API call in real implementation)
    weather_data = {
        "Seoul": {"temp": 22, "humidity": 55, "condition": "Clear", "wind": "3m/s"},
        "Tokyo": {"temp": 26, "humidity": 70, "condition": "Cloudy", "wind": "5m/s"},
        "New York": {"temp": 18, "humidity": 45, "condition": "Partly cloudy", "wind": "8m/s"},
    }
    data = weather_data.get(city, {"temp": 20, "humidity": 50, "condition": "No data", "wind": "N/A"})
    if unit == "fahrenheit":
        data["temp"] = round(data["temp"] * 9 / 5 + 32, 1)
    data["city"] = city
    data["unit"] = unit
    return data

def manage_schedule(action: str, date: str, **kwargs) -> dict:
    """Schedule management (in practice, connects to a DB or Google Calendar API)"""
    if action == "list":
        return {
            "date": date,
            "schedules": [
                {"id": "1", "title": "Team Meeting", "time": "10:00"},
                {"id": "2", "title": "Lunch Appointment", "time": "12:30"},
                {"id": "3", "title": "Code Review", "time": "15:00"},
            ]
        }
    elif action == "add":
        return {
            "status": "success",
            "message": f"Event '{kwargs.get('title', '')}' has been added on {date} {kwargs.get('time', '')}.",
            "id": "4"
        }
    elif action == "delete":
        return {
            "status": "success",
            "message": f"Schedule ID {kwargs.get('schedule_id', '')} has been deleted."
        }
    return {"status": "error", "message": "Unknown action."}

# ============================================
# 2. Tool name -> function mapping (dispatcher)
# ============================================
TOOL_FUNCTIONS = {
    "get_weather": get_weather,
    "manage_schedule": manage_schedule,
}

# ============================================
# 3. Agent loop (core code)
# ============================================
def run_agent(user_message: str, tools: list, max_turns: int = 10) -> str:
    """Agent loop that repeats tool calls until Claude gives a final answer"""
    messages = [{"role": "user", "content": user_message}]

    for turn in range(max_turns):
        # Call Claude API
        response = client.messages.create(
            model="claude-sonnet-4-5-20250514",
            max_tokens=4096,
            system="You are an assistant that helps with weather checks and schedule management. Respond naturally in English.",
            tools=tools,
            messages=messages,
        )

        # [Important] Add Claude's response to message history
        messages.append({"role": "assistant", "content": response.content})

        # If stop_reason is "end_turn," the final answer is complete
        if response.stop_reason == "end_turn":
            return "\n".join(
                block.text for block in response.content
                if hasattr(block, "text")
            )

        # If stop_reason is "tool_use," execute the tools
        if response.stop_reason == "tool_use":
            tool_results = []

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

                func = TOOL_FUNCTIONS.get(block.name)

                if func is None:
                    # Unknown tool: notify Claude with is_error=True
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,  # Must match!
                        "content": json.dumps({
                            "error": f"Tool '{block.name}' not found."
                        }),
                        "is_error": True,
                    })
                    continue

                try:
                    # Execute the tool
                    result = func(**block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result, ensure_ascii=False),
                    })
                except Exception as e:
                    # Execution error: is_error=True + error message
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps({
                            "error": f"{type(e).__name__}: {str(e)}"
                        }),
                        "is_error": True,
                    })

            # [Important] Add tool_results as a user message
            messages.append({"role": "user", "content": tool_results})

    return "Maximum conversation turns exceeded. Please try a more specific question."

# ============================================
# 4. Execution test
# ============================================
# Single tool call
result = run_agent("Tell me the weather in Seoul", tools)
print(result)
# -> "The current temperature in Seoul is 22 degrees with clear skies. Humidity is 55%..."

# Multiple tool calls (Claude decides automatically)
result = run_agent("Tell me the weather in Seoul and also check today's schedule", tools)
print(result)
# -> Claude calls both get_weather and manage_schedule

import anthropic
import json
import asyncio
from concurrent.futures import ThreadPoolExecutor

client = anthropic.Anthropic()

# ============================================
# Parallel tool execution (asyncio + ThreadPoolExecutor)
# ============================================
async def execute_tools_parallel(
    tool_blocks: list,
    tool_functions: dict,
    timeout_seconds: float = 30.0,
) -> list:
    """Execute multiple tools in parallel and collect results."""
    loop = asyncio.get_event_loop()
    executor = ThreadPoolExecutor(max_workers=5)

    async def run_single_tool(block):
        func = tool_functions.get(block.name)

        # 1. Unknown tool
        if not func:
            return {
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps({
                    "error": f"Tool '{block.name}' not found."
                }),
                "is_error": True,
            }

        try:
            # 2. Timeout + thread pool execution
            result = await asyncio.wait_for(
                loop.run_in_executor(
                    executor, lambda: func(**block.input)
                ),
                timeout=timeout_seconds,
            )
            return {
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps(result, ensure_ascii=False),
            }

        # 3. Timeout
        except asyncio.TimeoutError:
            return {
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps({
                    "error": f"Tool execution exceeded {timeout_seconds} seconds."
                }),
                "is_error": True,
            }

        # 4. Other errors (filter sensitive information)
        except Exception as e:
            return {
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps({
                    "error": f"Error during tool execution: {type(e).__name__}"
                    # str(e) may contain sensitive info, so only pass the type
                }),
                "is_error": True,
            }

    # Execute all tools in parallel
    tasks = [
        run_single_tool(b)
        for b in tool_blocks
        if b.type == "tool_use"
    ]
    results = await asyncio.gather(*tasks)
    return list(results)

# ============================================
# Production agent with API error handling
# ============================================
import time

def run_agent_production(
    user_message: str,
    tools: list,
    max_turns: int = 10,
    max_api_retries: int = 3,
) -> str:
    """Production-grade agent (error handling + parallel execution + retries)"""
    messages = [{"role": "user", "content": user_message}]

    for turn in range(max_turns):
        # API call (with retry logic)
        response = None
        for retry in range(max_api_retries):
            try:
                response = client.messages.create(
                    model="claude-sonnet-4-5-20250514",
                    max_tokens=4096,
                    system="You are an assistant that helps with weather checks and schedule management. Respond naturally in English.",
                    tools=tools,
                    messages=messages,
                )
                break  # Exit loop on success

            except anthropic.RateLimitError:
                # 429: Wait and retry
                wait_time = 2 ** retry  # 1s, 2s, 4s (exponential backoff)
                print(f"Rate limit exceeded. Waiting {wait_time} seconds before retrying...")
                time.sleep(wait_time)

            except anthropic.APIStatusError as e:
                if e.status_code >= 500:
                    # 5xx: Server error, retry
                    print(f"Server error ({e.status_code}). Retry {retry + 1}/{max_api_retries}")
                    time.sleep(1)
                else:
                    # 4xx (401, 403, etc.): Cannot retry
                    return f"API error: {e.status_code}. Please check your API key and settings."

            except anthropic.APIConnectionError:
                print(f"Network connection failed. Retry {retry + 1}/{max_api_retries}")
                time.sleep(2)

        if response is None:
            return "API call failed. Please check your network connection and API key."

        # Process response
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            return "\n".join(
                block.text for block in response.content
                if hasattr(block, "text")
            )

        if response.stop_reason == "tool_use":
            tool_blocks = [b for b in response.content if b.type == "tool_use"]
            # Parallel execution
            tool_results = asyncio.run(
                execute_tools_parallel(tool_blocks, TOOL_FUNCTIONS)
            )
            messages.append({"role": "user", "content": tool_results})

    return "Maximum conversation turns exceeded."

# ============================================
# Execution test: a case where parallel calls occur
# ============================================
result = run_agent_production(
    "Compare the weather in Seoul and Tokyo, and also check tomorrow's schedule",
    tools,
)
print(result)
# Claude calls get_weather("Seoul"), get_weather("Tokyo"),
# manage_schedule("list", "2025-05-16") - three tools in parallel

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

const client = new Anthropic();

// ============================================
// 1. Tool definitions (with TypeScript types)
// ============================================
const tools: Anthropic.Tool[] = [
  {
    name: "get_weather",
    description:
      "Retrieves current weather information (temperature, humidity, conditions) " +
      "for a specified city. City names can be in English or local language.",
    input_schema: {
      type: "object" as const,
      properties: {
        city: {
          type: "string",
          description: "City name (e.g., 'Seoul', 'Tokyo')",
        },
        unit: {
          type: "string",
          enum: ["celsius", "fahrenheit"],
          description: "Temperature unit (default: celsius)",
        },
      },
      required: ["city"],
    },
  },
  {
    name: "manage_schedule",
    description:
      "View (list), add, or delete user schedules.",
    input_schema: {
      type: "object" as const,
      properties: {
        action: {
          type: "string",
          enum: ["list", "add", "delete"],
          description: "Action to perform",
        },
        date: {
          type: "string",
          description: "Date (YYYY-MM-DD)",
        },
        title: {
          type: "string",
          description: "Event title (for add)",
        },
        time: {
          type: "string",
          description: "Time (HH:MM, for add)",
        },
        schedule_id: {
          type: "string",
          description: "Schedule ID (for delete)",
        },
      },
      required: ["action", "date"],
    },
  },
];

// ============================================
// 2. Tool execution functions (type-safe)
// ============================================
type ToolInput = Record<string, string>;

function executeTool(name: string, input: ToolInput): string {
  switch (name) {
    case "get_weather":
      // In real implementation, call a weather API
      return JSON.stringify({
        city: input.city,
        temp: 23,
        humidity: 60,
        condition: "Clear",
        wind: "4m/s",
        unit: input.unit || "celsius",
      });

    case "manage_schedule":
      if (input.action === "list") {
        return JSON.stringify({
          date: input.date,
          schedules: [
            { id: "1", title: "Team Standup", time: "09:30" },
            { id: "2", title: "Code Review", time: "14:00" },
            { id: "3", title: "1:1 Meeting", time: "16:00" },
          ],
        });
      }
      if (input.action === "add") {
        return JSON.stringify({
          status: "success",
          message: `Added '${input.title}' on ${input.date} ${input.time || ""}`,
          id: "4",
        });
      }
      if (input.action === "delete") {
        return JSON.stringify({
          status: "success",
          message: `Schedule ${input.schedule_id} deleted`,
        });
      }
      return JSON.stringify({ error: "Unknown action" });

    default:
      return JSON.stringify({ error: `Unknown tool: ${name}` });
  }
}

// ============================================
// 3. Agent main loop
// ============================================
async function runAgent(userMessage: string): Promise<string> {
  const messages: Anthropic.MessageParam[] = [
    { role: "user", content: userMessage },
  ];

  const MAX_TURNS = 10;

  for (let turn = 0; turn < MAX_TURNS; turn++) {
    const response = await client.messages.create({
      model: "claude-sonnet-4-5-20250514",
      max_tokens: 4096,
      system:
        "You are an assistant that helps with weather checks and schedule management. " +
        "Always respond naturally in English. " +
        "Summarize tool results in natural sentences rather than listing raw data.",
      tools,
      messages,
    });

    // Add assistant response to history
    messages.push({ role: "assistant", content: response.content });

    // If final answer, extract text and return
    if (response.stop_reason === "end_turn") {
      return response.content
        .filter((b): b is Anthropic.TextBlock => b.type === "text")
        .map((b) => b.text)
        .join("\n");
    }

    // Handle tool calls
    if (response.stop_reason === "tool_use") {
      const toolResults: Anthropic.ToolResultBlockParam[] =
        response.content
          .filter(
            (b): b is Anthropic.ToolUseBlock => b.type === "tool_use"
          )
          .map((block) => {
            try {
              const result = executeTool(
                block.name,
                block.input as ToolInput
              );
              return {
                type: "tool_result" as const,
                tool_use_id: block.id,
                content: result,
              };
            } catch (e) {
              return {
                type: "tool_result" as const,
                tool_use_id: block.id,
                content: JSON.stringify({
                  error:
                    e instanceof Error
                      ? e.message
                      : "Unknown execution error",
                }),
                is_error: true,
              };
            }
          });

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

  return "Maximum conversation turns exceeded.";
}

// ============================================
// 4. Execution (multi-step tool call scenario)
// ============================================
(async () => {
  const answer = await runAgent(
    "Tell me the weather in Seoul, check today's schedule, " +
    "and if the weather is nice, add a walk at 3 PM."
  );
  console.log(answer);
  // Claude processes in 3 steps:
  // 1. get_weather("Seoul") + manage_schedule("list", "today")
  // 2. Check weather -> if nice, manage_schedule("add", "today", "Walk", "15:00")
  // 3. Final summary response
})();

import anthropic
import json

client = anthropic.Anthropic()

# ============================================
# [Pattern 1] Managing tool result size
# ============================================
def search_database(query: str, limit: int = 10) -> dict:
    """If DB search results are large, summarize and return"""
    # Execute actual DB query
    all_results = [{"id": i, "name": f"Item {i}"} for i in range(100)]

    return {
        "total_count": len(all_results),
        "returned_count": min(limit, len(all_results)),
        "results": all_results[:limit],
        "message": f"Returning top {limit} of {len(all_results)} total results."
    }

# ============================================
# [Pattern 2] Tool input validation (security)
# ============================================
import re

def safe_file_read(filepath: str) -> dict:
    """Validate file path before reading"""
    # Allowed path list
    ALLOWED_DIRS = ["/data/reports/", "/data/exports/"]

    # Prevent path traversal attacks
    normalized = filepath.replace("\\", "/")
    if ".." in normalized:
        return {"error": "Invalid file path."}

    if not any(normalized.startswith(d) for d in ALLOWED_DIRS):
        return {"error": "Access to this path is not allowed."}

    # Actual file read (omitted)
    return {"content": "File contents...", "size": "1.2KB"}

# ============================================
# [Pattern 3] Controlling tool selection with tool_choice
# ============================================

# Default: Claude automatically decides whether to use tools
response_auto = client.messages.create(
    model="claude-sonnet-4-5-20250514",
    max_tokens=1024,
    tools=tools,
    tool_choice={"type": "auto"},  # Default
    messages=[{"role": "user", "content": "Hello!"}],
)
# -> Claude responds directly without tools (not weather/schedule related)

# Force a specific tool call
response_forced = client.messages.create(
    model="claude-sonnet-4-5-20250514",
    max_tokens=1024,
    tools=tools,
    tool_choice={"type": "tool", "name": "get_weather"},
    messages=[{"role": "user", "content": "Good morning!"}],
)
# -> Forces get_weather call (regardless of the question)

# Must use at least one tool
response_any = client.messages.create(
    model="claude-sonnet-4-5-20250514",
    max_tokens=1024,
    tools=tools,
    tool_choice={"type": "any"},
    messages=[{"role": "user", "content": "What should I do today?"}],
)
# -> Must call either weather or schedule tool

# ============================================
# [Pattern 4] Tool result caching (cost savings)
# ============================================
from functools import lru_cache
from datetime import datetime

@lru_cache(maxsize=100)
def get_weather_cached(city: str, date: str) -> str:
    """Cache same city+date combinations to prevent duplicate API calls"""
    # Actual weather API call (incurs cost)
    result = {"city": city, "temp": 22, "condition": "Clear"}
    return json.dumps(result, ensure_ascii=False)

# Usage: Even if the same city is queried multiple times on the same day, the API is called only once
today = datetime.now().strftime("%Y-%m-%d")
get_weather_cached("Seoul", today)  # API call
get_weather_cached("Seoul", today)  # Returned from cache (no API call)

# ============================================
# [Pattern 5] Logging and monitoring
# ============================================
import logging

logger = logging.getLogger("agent")

def run_agent_with_logging(user_message: str, tools: list) -> str:
    """Agent that logs all tool calls"""
    messages = [{"role": "user", "content": user_message}]
    total_input_tokens = 0
    total_output_tokens = 0

    for turn in range(10):
        response = client.messages.create(
            model="claude-sonnet-4-5-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        )

        # Track token usage
        total_input_tokens += response.usage.input_tokens
        total_output_tokens += response.usage.output_tokens

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

        if response.stop_reason == "end_turn":
            logger.info(
                f"Complete | Turns: {turn + 1} | "
                f"Input tokens: {total_input_tokens} | "
                f"Output tokens: {total_output_tokens} | "
                f"Est. cost: $" + f"{(total_input_tokens * 3 + total_output_tokens * 15) / 1_000_000:.4f}"
            )
            return "\n".join(
                b.text for b in response.content if hasattr(b, "text")
            )

        if response.stop_reason == "tool_use":
            for b in response.content:
                if b.type == "tool_use":
                    logger.info(f"Tool call: {b.name}({json.dumps(b.input)})")

            # Execute tools and return results (abbreviated)
            tool_results = []
            for b in response.content:
                if b.type == "tool_use":
                    func = TOOL_FUNCTIONS.get(b.name)
                    if func:
                        result = func(**b.input)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": b.id,
                            "content": json.dumps(result, ensure_ascii=False),
                        })

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

    return "Maximum turns exceeded"

Core Code

Core pattern of the Claude Tool Use agent loop: (1) define tools with JSON Schema, (2) include tools parameter in API call, (3) if stop_reason is "tool_use" execute the function, (4) return tool_result with tool_use_id matched exactly, (5) repeat until "end_turn".

import anthropic, json

client = anthropic.Anthropic()

# 1. Tool definition (JSON Schema)
tools = [{
    "name": "get_weather",
    "description": "Retrieves the current weather for a city",
    "input_schema": {
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "City name"}
        },
        "required": ["city"]
    }
}]

# 2. Agent loop (core pattern)
messages = [{"role": "user", "content": "What's the weather in Seoul?"}]

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

    # [Required] Add assistant response to history
    messages.append({"role": "assistant", "content": response.content})

    # If final answer, exit
    if response.stop_reason == "end_turn":
        print([b.text for b in response.content if hasattr(b, "text")])
        break

    # tool_use -> execute function -> return tool_result
    if response.stop_reason == "tool_use":
        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = execute_tool(block.name, block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,  # Must match!
                    "content": json.dumps(output, ensure_ascii=False),
                })
        messages.append({"role": "user", "content": results})

Common Mistakes

Related liminfo Services