Tool Use Patterns — Function Calling for LLMs

User: "What's the weather in Berlin?"
   ↓
LLM: "I need get_weather tool with city=Berlin"
   ↓
System: Calls get_weather("Berlin")
   ↓
Result: "20°C, sunny"
   ↓
LLM: "In Berlin it's 20°C and sunny"
   ↓
User: "Thanks!"

{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current weather for a location",
        "parameters": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "City name (e.g., Berlin, New York)"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "Temperature unit"
                }
            },
            "required": ["city"]
        }
    }
}

from anthropic import Anthropic

client = Anthropic()

tools = [
    {
        "type": "function",
        "function": {
            "name": "calculator",
            "description": "Perform math calculations",
            "parameters": {
                "type": "object",
                "properties": {
                    "operation": {
                        "type": "string",
                        "enum": ["add", "subtract", "multiply", "divide"]
                    },
                    "a": {"type": "number"},
                    "b": {"type": "number"}
                },
                "required": ["operation", "a", "b"]
            }
        }
    }
]

def execute_calculator(operation: str, a: float, b: float) -> str:
    if operation == "add":
        return str(a + b)
    elif operation == "multiply":
        return str(a * b)
    # ... etc

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

    while True:
        response = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )

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

        elif response.stop_reason == "tool_use":
            tool_results = []

            for content in response.content:
                if content.type == "tool_use":
                    result = execute_calculator(**content.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": content.id,
                        "content": result
                    })

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

solve_math("What is 123 * 456?")

# Claude calls get_stock_price AND get_news at same time
tool_calls = []

for content in response.content:
    if content.type == "tool_use":
        tool_calls.append(content)

# Execute all in parallel
results = []
for tool_call in tool_calls:
    result = execute_tool(tool_call.name, tool_call.input)
    results.append({
        "type": "tool_result",
        "tool_use_id": tool_call.id,
        "content": result
    })

# Send all results at once
messages.append({"role": "user", "content": results})

"description": "Tool"

"description": "Search products by price, category, and availability in our catalog"

"properties": {
    "q": {
        "type": "string",
        "description": "Search term (e.g., 'red shoes size 42')"
    }
}

"status": {
    "type": "string",
    "enum": ["pending", "processing", "completed"]
}

"max_results": {
    "type": "integer",
    "minimum": 1,
    "maximum": 100
}

class ProductionToolUse:
    def __init__(self, max_tool_calls: int = 10):
        self.client = Anthropic()
        self.max_tool_calls = max_tool_calls

    def execute_with_tools(self, prompt: str) -> str:
        messages = [{"role": "user", "content": prompt}]
        tool_call_count = 0

        while tool_call_count < self.max_tool_calls:
            response = self.client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1024,
                tools=self.tools,
                messages=messages
            )

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

            elif response.stop_reason == "tool_use":
                # Execute tools...
                tool_call_count += 1

        return None

class MultiTurnToolAgent:
    def __init__(self):
        self.tools = [
            self.search_tool,
            self.fetch_url_tool,
            self.parse_json_tool,
            self.save_results_tool
        ]

    def search_tool(self, query: str) -> str:
        """Search web for query."""
        return f"Found URLs for {query}: url1, url2, url3"

    def fetch_url_tool(self, url: str) -> str:
        """Fetch content from URL."""
        return f"Content from {url}: [mock content]"

    def parse_json_tool(self, text: str) -> dict:
        """Parse JSON from text."""
        return {"parsed": "data"}

    def save_results_tool(self, data: dict, filename: str) -> str:
        """Save results to file."""
        return f"Saved to {filename}"

    def run(self, task: str) -> str:
        """Example: 'Find AI news, parse articles, save results'"""
        messages = [{"role": "user", "content": task}]
        max_turns = 10

        for turn in range(max_turns):
            # Call Claude
            response = self.client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1024,
                tools=self._get_tools_schema(),
                messages=messages
            )

            if response.stop_reason == "end_turn":
                # Agent finished
                for block in response.content:
                    if hasattr(block, "text"):
                        return block.text

            # Agent wants to use tools
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    # Execute tool
                    result = self._execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })

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

        return "Max turns exceeded"

    def _execute_tool(self, name: str, input: dict) -> str:
        if name == "search":
            return self.search_tool(input["query"])
        elif name == "fetch_url":
            return self.fetch_url_tool(input["url"])
        elif name == "parse_json":
            return self.parse_json_tool(input["text"])
        elif name == "save_results":
            return self.save_results_tool(input["data"], input["filename"])

Scenario: "Get current Bitcoin price"

Without tool (make LLM guess):
  Input: 150 tokens
  Output: 100 tokens
  Total: 250 tokens
  Problem: Answer outdated (trained data cutoff)

With tool (use price API):
  Input: 150 tokens
  Tool schema: 50 tokens
  Output: 50 tokens (just "fetch current price")
  Tool result: 20 tokens (price data)
  Final output: 20 tokens
  Total: 290 tokens
  Problem solved: CURRENT price, not hallucinated

class ToolDiscoveryAgent:
    def __init__(self):
        self.available_tools = {
            "weather": "Get weather for city",
            "stocks": "Get stock price",
            "news": "Search news articles",
            "email": "Send email",
            "slack": "Post to Slack",
            "github": "Query GitHub repos",
            "database": "Query SQL database"
        }

    def solve(self, task: str) -> str:
        # Task: "What's happening with Tesla stock and relevant news?"
        # Agent figures out: needs stocks + news tools

        response = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            tools=[self._tool_to_schema(t) for t in self.available_tools.values()],
            messages=[{
                "role": "user",
                "content": f"Use available tools to: {task}"
            }]
        )

        # Agent picks 'stocks' and 'news' automatically
        # You don't tell it which tools to use—it discovers

class StreamingToolAgent:
    def run_with_streaming(self, task: str):
        messages = [{"role": "user", "content": task}]

        with self.client.messages.stream(
            model="claude-3-5-sonnet-20241022",
            tools=self.tools,
            messages=messages
        ) as stream:
            for text in stream.text_stream:
                # Tool decision arrives piece by piece
                print(text, end="", flush=True)

            # After stream, get tool uses
            for tool_use in stream.get_final_message().tool_use():
                result = self.execute_tool(tool_use.name, tool_use.input)
                # Continue conversation with result

Get user ID → Fetch user data → Send notification

# Claude decides:
# If stock price high: use alert tool
# If stock price low: use buy_recommendation tool

# Claude wants multiple data points at once:
# Get weather + stocks + news (all called in parallel)
results = await asyncio.gather(
    get_weather("Berlin"),
    get_stock_price("TSLA"),
    get_news("Tesla")
)

# WRONG: 50 tools, Claude confused
tools = [
    get_weather, get_stocks, get_news, get_email,
    send_message, create_file, delete_file, ...
    # (48 more)
]

# RIGHT: 5-10 focused tools
tools = [
    search_web,
    fetch_url,
    parse_json,
    save_file
]

# WRONG
"description": "Do something"

# RIGHT
"description": "Search recent news articles. Returns top 10 matches."

# WRONG
def send_email(to, body):
    # What if `to` is not valid email?
    smtp.send(to, body)

# RIGHT
def send_email(to: str, body: str) -> bool:
    if "@" not in to:
        raise ValueError(f"Invalid email: {to}")
    return smtp.send(to, body)

from functools import wraps
import time

class RateLimiter:
    def __init__(self, calls_per_second: int = 5):
        self.calls_per_second = calls_per_second
        self.last_call = 0

    def rate_limit(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - self.last_call
            min_interval = 1.0 / self.calls_per_second

            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)

            result = func(*args, **kwargs)
            self.last_call = time.time()
            return result

        return wrapper

limiter = RateLimiter(calls_per_second=3)

@limiter.rate_limit
def call_expensive_api():
    # Will never exceed 3 calls/second
    pass

class CostTracker:
    def __init__(self):
        self.tool_costs = {}
        self.total_cost = 0

    def track(self, tool_name: str, cost: float):
        if tool_name not in self.tool_costs:
            self.tool_costs[tool_name] = 0
        self.tool_costs[tool_name] += cost
        self.total_cost += cost

        if self.total_cost > 10:  # Max budget EUR 10
            raise RuntimeError(f"Budget exceeded: {self.total_cost}")

    def report(self):
        for tool, cost in sorted(self.tool_costs.items()):
            print(f"{tool}: €{cost:.4f}")
        print(f"Total: €{self.total_cost:.4f}")