gpt-5.5 (or any tool-calling OpenAI model) call Tako across turns until it has an answer.
Tako’s tool-calling primitives —
search, answer, and contents — are all fast, synchronous, stateless HTTP calls, which is exactly what function calling expects.Install
pip install openai tako-sdk
Set up your environment
You need an OpenAI key and a Tako API key:export OPENAI_API_KEY=<your openai key>
export TAKO_API_KEY=<your tako key>
TAKO_BASE_URL to target a non-prod host (e.g. https://staging.tako.com/api); leave it unset for production.
Define the tools
We register two clearly-differentiated tools. The descriptions are what make the model route correctly, so they spell out exactly when each applies:tako_answer— get a synthesized, written answer grounded in Tako cards and web results, plus the backing cards. This is the default for a data question.tako_contents— download the raw data (a CSV for a Tako card, extracted text for a web page) behind a result you already have, given that result’s URL.
TAKO_TOOLS = [
{
"type": "function",
"function": {
"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?')."
),
"parameters": {
"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"],
},
},
},
{
"type": "function",
"function": {
"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."
),
"parameters": {
"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."
)
Handle tool calls
When the model calls a tool, run it through the Tako SDK and return a JSON-serializable result. We summarize each card down to the fields a host agent actually needs to cite or embed it, and we return errors to the model rather than raising — so a failed call (for example, a transient network error) doesn’t kill the loop: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
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."""
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}"}
Run the agent loop
The loop is the standard OpenAI tool-calling pattern: call the model, and while it asks for tools, execute each call, append the result as atool message, and call again. When the model stops asking for tools, it has its answer:
from openai import OpenAI
def run_agent(openai_client: OpenAI, tako: Tako, user_message: str) -> str:
"""Run one user turn to completion, letting the model call Tako tools."""
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
]
while True:
completion = openai_client.chat.completions.create(
model="gpt-5.5", # swap for your preferred OpenAI model
messages=messages,
tools=TAKO_TOOLS,
)
message = completion.choices[0].message
messages.append(message)
if not message.tool_calls:
return message.content or ""
for tool_call in message.tool_calls:
args = json.loads(tool_call.function.arguments)
print(f" → {tool_call.function.name}({args})")
result = handle_tool_call(tako, tool_call.function.name, args)
messages.append(
{
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result),
}
)
Full code
The complete script below is also committed to the SDK repo atexamples/openai_tool_calling.py. Save it, set the two environment variables above, and run python openai_tool_calling.py.
"""Use Tako as a tool inside an OpenAI agent.
A copy-paste "Tako agent": an OpenAI chat 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 OPENAI_API_KEY=<your openai key>
uv run --group examples python examples/openai_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
from openai import OpenAI
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 (OpenAI shape) -----------------------------------------
# The descriptions are what make a host model pick the right tool, so they spell
# out exactly when to use each.
TAKO_TOOLS = [
{
"type": "function",
"function": {
"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?')."
),
"parameters": {
"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"],
},
},
},
{
"type": "function",
"function": {
"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."
),
"parameters": {
"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 as an `error` field 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(openai_client: OpenAI, tako: Tako, user_message: str) -> str:
"""Run one user turn to completion, letting the model call Tako tools."""
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_message},
]
while True:
completion = openai_client.chat.completions.create(
model="gpt-5.5", # swap for your preferred OpenAI model
messages=messages,
tools=TAKO_TOOLS,
)
message = completion.choices[0].message
messages.append(message)
if not message.tool_calls:
return message.content or ""
for tool_call in message.tool_calls:
args = json.loads(tool_call.function.arguments)
print(f" → {tool_call.function.name}({args})")
result = handle_tool_call(tako, tool_call.function.name, args)
messages.append(
{
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result),
}
)
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)
openai_client = OpenAI() # reads OPENAI_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(openai_client, tako_client, prompt)}")
Example output
Example output
User: What is the current US unemployment rate?
→ tako_answer({'query': 'current US unemployment rate', 'source_indexes': ['data']})
Assistant: The current US unemployment rate is 4.1%, per the "United States Unemployment Rate" card (latest monthly reading) — https://tako.com/embed/j60lxSe9mhx4-ni0ylTk
User: How has US unemployment trended over the last 5 years?
→ tako_answer({'query': 'US unemployment rate over the last 5 years'})
Assistant: It spiked to 14.8% in April 2020 during the COVID shock, fell back to roughly 3.5–4% through 2022–2023, and has hovered near 4% since. See the "United States Unemployment Rate" card — https://tako.com/embed/j60lxSe9mhx4-ni0ylTk
Instead of tako_answer, use tako_search
tako_answer returns prose plus the backing cards — best when you want the model to read an answer back to the user. If instead you want to render the cards yourself (embed the interactive charts, post-process the data) without a written answer on top, swap in tako_search, which returns the cards only.
Register this tool definition in place of tako_answer:
{
"type": "function",
"function": {
"name": "tako_search",
"description": (
"Search Tako's curated knowledge graph (and optionally the web) for data "
"on a topic. Returns ready-to-render knowledge cards — interactive charts "
"with their underlying data, plus title/description/embed_url — and web "
"results. Use when the user wants data, charts, statistics, trends, or "
"comparisons to display or post-process (e.g. 'Nvidia revenue since "
"2015'). Returns structured cards, NOT a written answer."
),
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Natural-language search query, e.g. 'Nvidia revenue since 2015'.",
},
"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 results per source (1-20). Default 5.",
},
},
"required": ["query"],
},
},
}
handle_tool_call:
if name == "tako_search":
response = tako.search(SearchRequest(query=args["query"], sources=_build_sources(args)))
return {"cards": _summarize_cards(response.cards)}
Register
tako_search or tako_answer, not both. They take the same inputs and differ only in whether you want prose or raw cards back, so exposing both forces the model to guess — weaker models route nearly every data question to tako_search. Pick the one that matches how your app consumes the result, and keep tako_contents alongside it.