"""Use Tako as a tool inside an Anthropic (Claude) agent.
A copy-paste "Tako agent": a Claude messages loop with two Tako tools —
tako_answer (a synthesized, grounded answer plus its backing cards) and
tako_contents (download the data behind a result you already have) — so the
model can answer data questions and fetch the underlying data on request.
Run:
export TAKO_API_KEY=<your tako key>
export ANTHROPIC_API_KEY=<your anthropic key>
uv run --group examples python examples/anthropic_tool_calling.py
Set TAKO_BASE_URL to target a non-prod host (e.g. https://staging.tako.com/api).
"""
from __future__ import annotations
import json
import os
import anthropic
from tako import Configuration
from tako.lib import Tako
from tako.models.contents_request import ContentsRequest
from tako.models.search_request import SearchRequest
from tako.models.source_settings import SourceSettings
from tako.models.sources import Sources
from tako.models.tako_source_settings import TakoSourceSettings
# --- Tool definitions (Anthropic shape) --------------------------------------
# Two tools in Anthropic's {name, description, input_schema} envelope. The
# descriptions are identical in wording to the OpenAI example: they are what
# make a host model route to the right tool.
TAKO_TOOLS = [
{
"name": "tako_answer",
"description": (
"Ask Tako a question and get a synthesized written answer grounded in Tako "
"knowledge cards and web results (returns the prose answer PLUS the backing "
"cards). Use when the user wants a direct answer to a factual or "
"quantitative question (e.g. 'What was US GDP growth last year?')."
),
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Natural-language question, e.g. 'What was US GDP growth last year?'.",
},
"source_indexes": {
"type": "array",
"items": {"type": "string", "enum": ["data", "web"]},
"description": (
"Where to search. Omit to use both Tako's curated cards and the live web (default); "
"['data'] = curated cards only, ['web'] = web only. "
"The legacy value 'tako' is accepted as a synonym for 'data'."
),
},
"count": {
"type": "integer",
"minimum": 1,
"maximum": 20,
"description": "Max backing results per source (1-20). Default 5.",
},
},
"required": ["query"],
},
},
{
"name": "tako_contents",
"description": (
"Download the content behind a result you already have, given THAT "
"result's URL. A Tako card URL returns a CSV of its underlying dataset; a "
"web URL returns the page's extracted full text. Use ONLY after a prior "
"tako_answer surfaced a result you want the raw data/text for (e.g. the "
"user says 'download that data'). The input is a result URL, NOT a search "
"query."
),
"input_schema": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": (
"The result URL to download (a Tako card's webpage_url or a "
"web result's url)."
),
},
},
"required": ["url"],
},
},
]
SYSTEM_PROMPT = (
"You are a data analyst assistant with access to Tako. Use tako_answer to "
"answer the user's data questions, and tako_contents to download the data "
"behind a result when asked. Cite card titles and include their embed_url links."
)
def _summarize_cards(cards) -> list:
return [
{
"card_id": card.card_id,
"title": card.title,
"description": card.description,
"embed_url": card.embed_url,
"webpage_url": card.webpage_url,
}
for card in (cards or [])
]
def _build_sources(args: dict) -> Sources | None:
"""Translate the model's `source_indexes` + `count` into the per-source `sources` object.
A source is searched only if its key is present. Returns None to use the
default (both Tako and web, 5 results each).
"""
indexes = args.get("source_indexes")
count = args.get("count")
if not indexes and count is None:
return None
settings = {"count": count} if count is not None else {}
indexes = indexes or ["data", "web"]
return Sources(
data=TakoSourceSettings(**settings) if ("data" in indexes or "tako" in indexes) else None,
web=SourceSettings(**settings) if "web" in indexes else None,
)
def handle_tool_call(tako: Tako, name: str, args: dict) -> dict:
"""Execute one Tako tool call and return a JSON-serializable result.
Errors are returned to the model rather than raised, so the loop keeps
going.
"""
try:
if name == "tako_answer":
response = tako.answer(SearchRequest(query=args["query"], sources=_build_sources(args)))
return {"answer": response.answer, "cards": _summarize_cards(response.cards)}
if name == "tako_contents":
response = tako.contents(ContentsRequest(url=args["url"]))
return {
"contents": [
{"format": item.format.value, "url": item.url, "expires_at": item.expires_at}
for item in (response.contents or [])
]
}
return {"error": f"unknown tool {name}"}
except Exception as exc:
return {"error": f"{name} failed: {exc}"}
def run_agent(anthropic_client: anthropic.Anthropic, tako: Tako, user_message: str) -> str:
"""Run one user turn to completion, letting Claude call Tako tools."""
messages = [{"role": "user", "content": user_message}]
while True:
response = anthropic_client.messages.create(
model="claude-opus-4-8",
max_tokens=1024,
system=SYSTEM_PROMPT,
tools=TAKO_TOOLS,
messages=messages,
)
if response.stop_reason != "tool_use":
return "".join(block.text for block in response.content if block.type == "text")
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
print(f" → {block.name}({block.input})")
result = handle_tool_call(tako, block.name, dict(block.input))
tool_results.append(
{
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result),
}
)
messages.append({"role": "user", "content": tool_results})
if __name__ == "__main__":
config = Configuration(host=os.environ.get("TAKO_BASE_URL")) # unset -> prod default
config.api_key["apiKey"] = os.environ["TAKO_API_KEY"]
tako_client = Tako(config)
claude = anthropic.Anthropic() # reads ANTHROPIC_API_KEY
for prompt in [
"What is the current US unemployment rate?",
"How has US unemployment trended over the last 5 years?",
]:
print(f"\nUser: {prompt}")
print(f"Assistant: {run_agent(claude, tako_client, prompt)}")
