claude 3.7 thinking tool usage example

claude_tool_usage.py

import os
import json
import logging
import pprint
from typing import List, Dict, Any, Optional, Literal
from anthropic import Anthropic
from dotenv import load_dotenv

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

# Get API key from environment
api_key = os.environ.get("ANTHROPIC_API_KEY")
if not api_key:
    raise ValueError("ANTHROPIC_API_KEY environment variable not set")

# Initialize Anthropic client
client = Anthropic(api_key=api_key)

def save_file_tool(file_path: str, content: str) -> Dict[str, Any]:
    """Tool for saving content to a file."""
    try:
        # Ensure file_path is not empty
        if not file_path:
            return {
                "status": "error",
                "message": "Empty file path provided",
            }
            
        # Make sure directory exists
        directory = os.path.dirname(file_path)
        if directory and not os.path.exists(directory):
            os.makedirs(directory, exist_ok=True)
            
        # Write the file
        with open(file_path, 'w') as f:
            f.write(content)
            
        return {
            "status": "success",
            "message": f"File saved successfully to {file_path}",
        }
    except Exception as e:
        return {
            "status": "error",
            "message": f"Error saving file: {str(e)}",
        }

def run_terminal_command_tool(command: str) -> Dict[str, Any]:
    """Tool for running terminal commands."""
    import subprocess
    try:
        result = subprocess.run(
            command, 
            shell=True, 
            capture_output=True, 
            text=True,
            timeout=300,  # 5 minute timeout
        )
        return {
            "status": "success" if result.returncode == 0 else "error",
            "stdout": result.stdout,
            "stderr": result.stderr,
            "returncode": result.returncode,
        }
    except subprocess.TimeoutExpired:
        return {
            "status": "error",
            "message": "Command timed out after 5 minutes",
        }
    except Exception as e:
        return {
            "status": "error",
            "message": f"Error executing command: {str(e)}",
        }

# Define tools
tools = [
    {
        "name": "save_file",
        "description": "Save content to a file. Creates directories in the path if they don't exist.",
        "input_schema": {
            "type": "object",
            "properties": {
                "file_path": {
                    "type": "string",
                    "description": "Path where the file should be saved"
                },
                "content": {
                    "type": "string",
                    "description": "Content to write to the file"
                }
            },
            "required": ["file_path", "content"]
        }
    },
    {
        "name": "run_terminal_command",
        "description": "Run a terminal command and return the result.",
        "input_schema": {
            "type": "object",
            "properties": {
                "command": {
                    "type": "string",
                    "description": "Command to execute in the terminal"
                }
            },
            "required": ["command"]
        }
    }
]

def process_tool_calls(tool_calls: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Process tool calls from Claude and execute them."""
    tool_results = []
    
    for tool_call in tool_calls:
        tool_name = tool_call.get("name")
        tool_input = tool_call.get("input", {})
        tool_id = tool_call.get("id")
        
        logger.info(f"Processing tool call: {tool_name} with input: {tool_input}")
        
        result = None
        if tool_name == "save_file":
            file_path = tool_input.get("file_path", "")
            content = tool_input.get("content", "")
            result = save_file_tool(file_path, content)
        elif tool_name == "run_terminal_command":
            command = tool_input.get("command", "")
            result = run_terminal_command_tool(command)
        else:
            result = {
                "status": "error",
                "message": f"Unknown tool: {tool_name}"
            }
        
        # Convert result to JSON string if it's not already a string
        result_str = result if isinstance(result, str) else json.dumps(result)
        logger.info(f"Tool call result: {result}")
        
        tool_results.append({
            "tool_use_id": tool_id,
            "content": result_str
        })
    
    return tool_results

def extract_response_text(response):
    """Extract text from response safely."""
    if hasattr(response, 'content') and response.content:
        for content_block in response.content:
            if hasattr(content_block, 'text'):
                return content_block.text
    return "No text in response"

def get_tool_use(response):
    """Extract tool use from response safely."""
    try:
        if hasattr(response, 'model_dump'):
            response_dict = response.model_dump()
            
            # Look for tool_use blocks in content
            content = response_dict.get('content', [])
            tool_calls = []
            
            for block in content:
                if block.get('type') == 'tool_use':
                    # Create a tool call object with the expected format
                    tool_call = {
                        'id': block.get('id'),
                        'name': block.get('name'),
                        'input': block.get('input', {})
                    }
                    tool_calls.append(tool_call)
            
            if tool_calls:
                return tool_calls
            
            # If stop_reason is tool_use but we didn't find tool calls in content,
            # something unexpected happened
            if response_dict.get('stop_reason') == 'tool_use':
                logger.warning("Response has stop_reason='tool_use' but no tool calls were extracted")
    except Exception as e:
        logger.error(f"Error extracting tool calls: {e}")
    
    return []

def test_multi_stage_tool_calling():
    """Test a multi-stage tool calling process with Claude 3.7 with thinking model."""
    
    # Initial user request
    user_request = "I want to visualize the stock prices of Google and Facebook from 2024 on the same chart."
    
    # Set up conversation history
    conversation = [{"role": "user", "content": user_request}]
    
    # Multi-stage conversation loop
    max_iterations = 5
    iteration = 0
    while iteration < max_iterations:
        logger.info(f"Iteration {iteration+1}/{max_iterations}")
        
        # Send message to Claude
        logger.info(f"Sending request to Claude with {len(conversation)} messages")
        response = client.beta.messages.create(
            model="claude-3-7-sonnet-20250219",
            max_tokens=4000,
            thinking={
                "type": "enabled",
                "budget_tokens": 2000
            },
            messages=conversation,
            tools=tools,
        )
        
        # Print debug information about the response
        print("\n" + "="*80)
        print(f"CLAUDE RESPONSE (Iteration {iteration+1}):")
        print("="*80)
        
        # Extract and print response text for debugging
        response_text = extract_response_text(response)
        print("\nResponse text:")
        print(response_text)
        
        # Extract response content blocks to preserve tool_use blocks
        response_dict = response.model_dump()
        response_content = response_dict.get('content', [])
        
        # Add Claude's complete response to conversation (with tool_use blocks preserved)
        conversation.append({"role": "assistant", "content": response_content})
        
        # Extract tool use information
        tool_calls = get_tool_use(response)
        if tool_calls:
            logger.info(f"Claude wants to use {len(tool_calls)} tools")
            print("\nTool calls:")
            for tool_call in tool_calls:
                print(f"- {tool_call['name']}: {tool_call['input']}")
            
            # Process tool calls
            tool_results = process_tool_calls(tool_calls)
            
            # Format tool results as content blocks in a user message
            tool_result_blocks = []
            for result in tool_results:
                tool_result_blocks.append({
                    "type": "tool_result",
                    "tool_use_id": result["tool_use_id"],
                    "content": result["content"]
                })
            
            # Add a user message with tool result content blocks
            conversation.append({
                "role": "user", 
                "content": tool_result_blocks
            })
            
            # Debug the tool_results
            print("\nTool results:")
            for result in tool_results:
                print(f"- {result['tool_use_id']}: {result['content']}")
        else:
            # If no tool calls, we're done with this iteration
            logger.info("No tool calls in this response")
            print("\nNo tool calls in this response")
            break
        
        # Move to the next iteration
        iteration += 1
    
    return response

if __name__ == "__main__":
    logger.info("Starting tool calling test with Claude 3.7")
    response = test_multi_stage_tool_calling()
    logger.info("Completed tool calling test with Claude 3.7")