Tool Use and Function Calling: Letting LLMs Query Data, Call APIs and Take Actions

import json, re, duckdb, pandas as pd, numpy as np
sales = pd.read_csv("superstore_sales.csv", parse_dates=["order_date"])
con = duckdb.connect(); con.register("sales", sales)
TOOLS = [
    {"name": "run_sql",
     "description": "Run a read-only DuckDB SQL query on table `sales` (one row per order line) and return up to 20 rows as JSON. "
                    "Columns: order_id, order_date (DATE), ship_date, ship_mode, customer_id, segment, city, state, region, category, sub_category, "
                    "product_name, sales (INR), quantity, discount (0-1), profit (INR). Use this for any question about sales data.",
     "input_schema": {"type": "object", "properties": {"sql": {"type": "string", "description": "A single SELECT statement"}}, "required": ["sql"], "additionalProperties": False}},
    {"name": "calculator",
     "description": "Evaluate an arithmetic expression exactly, e.g. '(78.38 - 65.40) / 65.40 * 100'. Use for growth rates and percentages.",
     "input_schema": {"type": "object", "properties": {"expression": {"type": "string"}}, "required": ["expression"], "additionalProperties": False}},
    {"name": "convert_currency",
     "description": "Convert an amount from INR to USD, EUR or AED using today's reference rate.",
     "input_schema": {"type": "object", "properties": {"amount_inr": {"type": "number"}, "to": {"type": "string", "enum": ["USD", "EUR", "AED"]}}, "required": ["amount_inr", "to"], "additionalProperties": False}},
]
print(json.dumps(TOOLS[1], indent=1))

import ast, operator as op
def run_sql(sql: str):
    if not re.match(r"^\s*(with|select)\b", sql, re.I) or re.search(r"\b(insert|update|delete|drop|alter|create|copy|attach|pragma)\b", sql, re.I):
        raise ValueError("only read-only SELECT queries are allowed")
    return con.execute(sql).df().head(20).round(2).to_dict("records")
OPS = {ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv, ast.Pow: op.pow, ast.USub: op.neg}
def calculator(expression: str):                     # safe arithmetic: no eval()
    def ev(n):
        if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
        if isinstance(n, ast.BinOp) and type(n.op) in OPS: return OPS[type(n.op)](ev(n.left), ev(n.right))
        if isinstance(n, ast.UnaryOp) and type(n.op) in OPS: return OPS[type(n.op)](ev(n.operand))
        raise ValueError("unsupported expression")
    return round(ev(ast.parse(expression, mode="eval").body), 4)
RATES = {"USD": 1 / 88.0, "EUR": 1 / 102.0, "AED": 1 / 24.0}      # illustrative reference rates
def convert_currency(amount_inr: float, to: str): return {"amount": round(amount_inr * RATES[to], 2), "currency": to}
REGISTRY = {"run_sql": run_sql, "calculator": calculator, "convert_currency": convert_currency}
def execute_tool(name, tool_input):
    try: return {"content": json.dumps(REGISTRY[name](**tool_input), default=str)}
    except Exception as e: return {"content": f"Error: {e}", "is_error": True}       # errors go back to the model, which can fix its call
print(execute_tool("run_sql", {"sql": "SELECT region, ROUND(SUM(sales)/1e6, 2) AS sales_m FROM sales GROUP BY region ORDER BY sales_m DESC"}))
print(execute_tool("run_sql", {"sql": "DROP TABLE sales"}))
print(execute_tool("calculator", {"expression": "__import__('os').system('ls')"}))

import types
def block(**kw): return types.SimpleNamespace(**kw)
class ScriptedModel:
    """Emits Anthropic-style responses: tool_use blocks first, then a final text answer that uses the tool results."""
    def __init__(self, plan): self.plan, self.step = plan, 0
    def create(self, messages, tools, **kw):
        if self.step < len(self.plan):
            calls = self.plan[self.step](messages); self.step += 1
            return block(stop_reason="tool_use", content=[block(type="text", text="Let me check the data.")] +
                         [block(type="tool_use", id=f"toolu_{self.step}_{i}", name=n, input=a) for i, (n, a) in enumerate(calls)])
        results = [json.loads(c["content"]) for m in messages if m["role"] == "user" and isinstance(m["content"], list) for c in m["content"] if c.get("type") == "tool_result" and not c.get("is_error")]
        return block(stop_reason="end_turn", content=[block(type="text", text=FINAL(results))])

question = "How much did 2025 sales grow over 2024, and what is 2025 sales in US dollars?"
plan = [lambda m: [("run_sql", {"sql": "SELECT YEAR(order_date) AS yr, ROUND(SUM(sales)/1e6, 2) AS sales_m FROM sales WHERE YEAR(order_date) IN (2024, 2025) GROUP BY yr ORDER BY yr"})],
        lambda m: (lambda r: [("calculator", {"expression": f"({r[1]['sales_m']} - {r[0]['sales_m']}) / {r[0]['sales_m']} * 100"}),          # two tool calls in parallel
                              ("convert_currency", {"amount_inr": r[1]["sales_m"] * 1e6, "to": "USD"})])(json.loads(m[-1]["content"][0]["content"]))]
FINAL = lambda res: (f"2025 sales were ₹{res[0][1]['sales_m']}M against ₹{res[0][0]['sales_m']}M in 2024, growth of {res[1]:.1f}%. "
                     f"That is about ${res[2]['amount'] / 1e6:.2f}M at the reference rate.")
model = ScriptedModel(plan)
messages = [{"role": "user", "content": question}]
while True:
    resp = model.create(messages=messages, tools=TOOLS)
    messages.append({"role": "assistant", "content": resp.content})
    if resp.stop_reason != "tool_use": break
    results = []
    for b in resp.content:
        if b.type == "tool_use":
            out = execute_tool(b.name, b.input); print(f"→ {b.name}({json.dumps(b.input)[:90]}) = {out['content'][:90]}")
            results.append({"type": "tool_result", "tool_use_id": b.id, **out})
    messages.append({"role": "user", "content": results})                 # ALL results of a turn go back in ONE user message
print("\nAnswer:", next(b.text for b in resp.content if b.type == "text"))

import anthropic
client = anthropic.Anthropic()
messages = [{"role": "user", "content": question}]
while True:
    response = client.messages.create(model="claude-opus-5-5", max_tokens=16000, tools=TOOLS, messages=messages)
    messages.append({"role": "assistant", "content": response.content})      # keep the full content, including tool_use blocks
    if response.stop_reason != "tool_use":
        break
    results = []
    for b in response.content:
        if b.type == "tool_use":
            results.append({"type": "tool_result", "tool_use_id": b.id, **execute_tool(b.name, b.input)})
    messages.append({"role": "user", "content": results})
print(next(b.text for b in response.content if b.type == "text"))

from anthropic import beta_tool

@beta_tool
def calculator(expression: str) -> str:
    """Evaluate an arithmetic expression exactly.

    Args:
        expression: e.g. '(78.38 - 65.40) / 65.40 * 100'
    """
    return str(safe_eval(expression))

runner = client.beta.messages.tool_runner(model="claude-opus-5-5", max_tokens=16000, tools=[calculator, run_sql_tool],
                                          messages=[{"role": "user", "content": question}])
for message in runner:          # iterates until Claude stops calling tools
    print(message.content[-1])

response = client.messages.create(
    model="claude-opus-5-5", max_tokens=16000,
    tools=[{"type": "web_search_20260209", "name": "web_search", "max_uses": 3, "allowed_domains": ["rbi.org.in", "pib.gov.in"]}],
    messages=[{"role": "user", "content": "What is the current RBI repo rate? Cite the source."}],
)