Anthropic Tool Use with a Live Search Backend
The raw Messages API tool loop, done properly: schema design, the stop_reason dance, parallel tool calls, and where the loop should terminate.
Frameworks hide the tool loop, which is fine until something goes wrong and you have no idea what the model actually received. The raw loop is about forty lines. Worth writing once.
Tool schemas
import os
import json
import requests
from urllib.parse import quote_plus
import anthropic
SERPLY_KEY = os.environ["SERPLY_API_KEY"]
BASE = "https://api.serply.io/v1"
TOOLS = [
{
"name": "web_search",
"description": (
"Search the live web and return ranked results with titles, URLs, and "
"snippets. Use this for current events, prices, product details, or any "
"fact that may have changed since your training data. Prefer searching "
"over guessing."
),
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query, phrased as you'd type it into Google.",
},
"num": {
"type": "integer",
"description": "Number of results to return.",
"default": 10,
},
},
"required": ["query"],
},
},
{
"name": "read_page",
"description": (
"Fetch the full text of a web page as markdown. Use after web_search "
"when a snippet doesn't contain enough detail to answer confidently."
),
"input_schema": {
"type": "object",
"properties": {
"url": {"type": "string", "description": "Full URL of the page."},
},
"required": ["url"],
},
},
]“Prefer searching over guessing” earns its place. Claude is well-calibrated about uncertainty but defaults to answering from memory when a question feels answerable; an explicit nudge in the tool description shifts that materially.
The implementations
def web_search(query: str, num: int = 10) -> str:
resp = requests.get(
f"{BASE}/search/q={quote_plus(query)}&num={num}",
headers={"X-Api-Key": SERPLY_KEY, "X-Proxy-Location": "US"},
timeout=30,
)
if resp.status_code == 429:
return "Rate limited. Do not retry; answer with available information."
resp.raise_for_status()
results = resp.json().get("results", [])
if not results:
return f"No results for '{query}'."
return "\n\n".join(
f"[{r.get('position')}] {r.get('title')}\n{r.get('link')}\n{r.get('description', '')}"
for r in results
)
def read_page(url: str) -> str:
resp = requests.post(
f"{BASE}/request",
headers={"X-Api-Key": SERPLY_KEY, "Content-Type": "application/json"},
json={"url": url, "response_type": "markdown"},
timeout=60,
)
if resp.status_code == 429:
return "Rate limited while reading that page."
resp.raise_for_status()
return resp.text[:15000]
DISPATCH = {"web_search": web_search, "read_page": read_page}Serply’s search endpoint puts the query string in the path (/v1/search/q=...). The scraper’s markdown mode returns text as the body — .text, not .json() — while "response_type": "full" returns JSON with the raw HTML under data.
The loop
client = anthropic.Anthropic()
SYSTEM = (
"You answer with live information. Search before making factual claims about "
"anything current. Cite every claim with its source URL. If searches don't "
"support an answer, say so rather than filling the gap from memory."
)
def run(question: str, max_turns: int = 8) -> str:
messages = [{"role": "user", "content": question}]
for _ in range(max_turns):
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=4096,
system=SYSTEM,
tools=TOOLS,
messages=messages,
)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
return "".join(
b.text for b in response.content if b.type == "text"
)
results = []
for block in response.content:
if block.type != "tool_use":
continue
try:
output = DISPATCH[block.name](**block.input)
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": output,
})
except Exception as e:
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": f"Tool failed: {e}",
"is_error": True,
})
messages.append({"role": "user", "content": results})
return "Stopped after the maximum number of tool turns."Three things this gets right that hand-rolled loops usually get wrong.
Every tool_use block in a response gets a matching tool_result in the next message. Claude issues parallel tool calls freely — three searches in one turn is common — and the API rejects the next request if any tool_use_id goes unanswered.
Errors come back as tool_result with is_error: true, not as exceptions that kill the run. The model reads them and adapts, which is usually better than crashing.
The turn cap is a hard number, not a model judgment. Without it, a question with no good answer on the open web will loop until something else breaks.
Running tools in parallel
Since Claude batches tool calls, run them concurrently:
from concurrent.futures import ThreadPoolExecutor
blocks = [b for b in response.content if b.type == "tool_use"]
with ThreadPoolExecutor(max_workers=min(len(blocks), 5)) as pool:
outputs = list(pool.map(
lambda b: DISPATCH[b.name](**b.input), blocks
))
results = [
{"type": "tool_result", "tool_use_id": b.id, "content": o}
for b, o in zip(blocks, outputs)
]Three sequential page reads at roughly two seconds each is six seconds of dead time per turn. Concurrent, it’s two. On a multi-turn research question that difference compounds into something users notice.
Prompt caching
Tool definitions and a long system prompt get resent on every turn. Mark them cacheable and the repeated cost largely disappears:
system=[{"type": "text", "text": SYSTEM, "cache_control": {"type": "ephemeral"}}],On an eight-turn research loop this is the single highest-leverage change you can make to the bill.