Tako Python SDK

The Tako Python SDK provides convenient access to the Tako API from any Python 3.9+ application. It ships fully typed request and response models and offers both synchronous and asynchronous clients.

Documentation

API reference and guides are available at docs.tako.com.

Installation

pip install tako-sdk

The import package is tako:

from tako.lib import Tako

Authentication

Create an API key from your Tako account and provide it when building the client. We recommend keeping it out of source control — for example, reading it from an environment variable:

import os

from tako import Configuration
from tako.lib import Tako

config = Configuration()
config.api_key["apiKey"] = os.environ["TAKO_API_KEY"]
client = Tako(config)

Usage

import os

from tako import Configuration, SearchRequest
from tako.lib import Tako

config = Configuration()
config.api_key["apiKey"] = os.environ["TAKO_API_KEY"]
client = Tako(config)

results = client.search(SearchRequest(query="S&P 500 performance this year"))
print(results.request_id)
for card in results.cards or []:
    print(card.title, card.webpage_url)

Operations

Method	Description
`client.search(SearchRequest(...))`	Search the Tako knowledge base; returns matching cards and web results.
`client.answer(SearchRequest(...))`	Get a written answer with supporting cards.
`client.create_card(CreateCardRequest(...))`	Build a visualization card from component configurations.
`client.contents(ContentsRequest(...))`	Fetch downloadable content (e.g. a CSV) for a card or web URL.

For example, fetch the underlying data for a card returned by a search. Not every card is exportable (some come from protected sources), so guard for the case where no card has downloadable content:

from tako import ContentsRequest, SearchRequest

results = client.search(SearchRequest(query="US Oil Prices"))
card = next(
    (c for c in (results.cards or []) if c.webpage_url and c.content and c.content.formats),
    None,
)
if card is None:
    print("No exportable card found")
else:
    contents = client.contents(ContentsRequest(url=card.webpage_url))
    for item in contents.contents or []:
        print(item.content_format, item.url)

Async usage

Use AsyncTako with the tako.aio package and await each call:

import asyncio
import os

from tako.aio import Configuration, SearchRequest
from tako.lib import AsyncTako


async def main() -> None:
    config = Configuration()
    config.api_key["apiKey"] = os.environ["TAKO_API_KEY"]
    client = AsyncTako(config)

    results = await client.search(SearchRequest(query="S&P 500 performance this year"))
    print(results.request_id)


asyncio.run(main())

The async client exposes the same operations as the synchronous one.

Agents

Two agent products hang off client.agent:

Namespace	Endpoint	Product
`client.agent.retrieval.*`	`/v1/agent/retrieval/runs`	Retrieval Agent — agentic data retrieval (multi-hop lookup, cohort resolution, structured outputs)
`client.agent.answer.*`	`/v1/agent/answer/runs`	Answer Agent — opinionated agentic research returning cited prose

Each exposes run(req) (202 dispatch → run object), get(run_id) (poll for status), and stream(req) (live SSE). (client.agent.answer.*, the Answer Agent, is distinct from client.answer(), the one-shot /v1/answer call.)

Streaming

Stream a run live over Server-Sent Events. The stream yields typed per-product envelopes (RetrievalAgentStreamEnvelope / AnswerAgentStreamEnvelope) and auto-reconnects (resuming via the last seq) on transient network drops. Use it as a context manager so the connection is always closed.

from tako import Configuration
from tako.lib import Tako
from tako.models.retrieval_agent_run_request import RetrievalAgentRunRequest

config = Configuration()
config.api_key["apiKey"] = "YOUR_API_KEY"
client = Tako(config)

req = RetrievalAgentRunRequest(query="Which S&P 500 semis grew revenue fastest in 2024?")
with client.agent.retrieval.stream(req) as stream:
    for event in stream:
        block = event.block.actual_instance
        print(event.seq, block.kind)
    # The stream ends at `stream_done`. If it ended without a terminal result
    # (and produced at least one event, so `run_id` is known), poll for status:
    if stream.result is None and stream.run_id is not None:
        run = client.agent.retrieval.get(stream.run_id)
        print(run.status)

Async usage mirrors this — stream = await client.agent.retrieval.stream(req) then async with stream: async for event in stream: .... The Answer Agent is identical with client.agent.answer.* and AnswerAgentRunRequest.

Structured output (Retrieval Agent)

Pass an output_schema (JSON Schema) to shape the response. Mark a property with "x-tako-dataset": true to request a dataset slot — filled with exact retrieved rows as a TakoDataset. Two helpers make this ergonomic:

derive_response_schema(schema) — the schema structured_output actually validates against (each slot becomes TakoDataset | null). Pair with jsonschema.
TakoDatasetView — a records / DataFrame view over a filled slot. .records needs no extra dependency; .to_dataframe() needs the pandas extra (pip install tako-sdk[pandas]).

import jsonschema
from tako.lib import Tako, TakoDatasetView, derive_response_schema
from tako.models.retrieval_agent_run_request import RetrievalAgentRunRequest

schema = {
    "type": "object",
    "properties": {
        "headline": {"type": "string"},
        "cohort": {"x-tako-dataset": True, "columns": ["company", "revenue"]},
    },
    "required": ["headline", "cohort"],
}

run = client.agent.retrieval.run(RetrievalAgentRunRequest(query="...", output_schema=schema))
run = client.agent.retrieval.get(run.run_id)  # poll to a terminal status

if run.result and run.result.structured_output:
    jsonschema.validate(run.result.structured_output, derive_response_schema(schema))
    view = TakoDatasetView(run.result.structured_output["cohort"])
    print(view.records)          # list[dict], one per row
    print(view.to_dataframe())   # typed pandas DataFrame (needs tako-sdk[pandas])

See examples/ for runnable scripts — sync (retrieval_agent_streaming.py, answer_agent_streaming.py, retrieval_agent_structured_output.py) and async (retrieval_agent_streaming_async.py, answer_agent_streaming_async.py).

Requests and responses

Request and response models are Pydantic models. Access fields as attributes (results.request_id), and use the usual helpers to serialize:

results.model_dump()        # -> dict
results.model_dump_json()   # -> JSON string

Configuration

By default the client targets the Tako production API. To point at a different host, pass it to Configuration:

from tako import Configuration

config = Configuration(host="https://staging.tako.com/api")

Handling errors

API errors raise a subclass of tako.ApiException. The exception carries the HTTP status, reason, and response body:

from tako import Configuration, SearchRequest
from tako.exceptions import ApiException, UnauthorizedException
from tako.lib import Tako

config = Configuration()
config.api_key["apiKey"] = "invalid-key"
client = Tako(config)

try:
    client.search(SearchRequest(query="US GDP growth rate"))
except UnauthorizedException:
    print("Invalid or missing API key")
except ApiException as exc:
    print(f"Request failed: {exc.status} {exc.reason}")
    print(exc.body)

The async client (AsyncTako) raises the same exception classes, so you can catch tako.exceptions.ApiException around await calls exactly as above — no async-specific imports are needed.

Status codes map to the following exception types (all subclasses of ApiException):

Status Code	Exception
400	`BadRequestException`
401	`UnauthorizedException`
403	`ForbiddenException`
404	`NotFoundException`
409	`ConflictException`
422	`UnprocessableEntityException`
>=500	`ServiceException`

Versioning

This package follows SemVer. You can check the installed version at runtime:

import tako

print(tako.__version__)

Requirements

Python 3.9 or higher.

Support

Questions, bugs, or feedback? See the documentation at docs.tako.com.

Release files for tako-sdk 2.2.20

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for tako-sdk 2.2.20
File	Size	Uploaded
tako_sdk-2.2.20.tar.gz	163.0 kB	Aug 25, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for tako-sdk 2.2.20
File	Interpreter	ABI	Platform	Reset
tako_sdk-2.2.20-py3-none-any.whl 380.1 kB Aug 25, 2026	Python 3	none	any	Details

Total release size: 543.1 kB

Release files / tako_sdk-2.2.20.tar.gz

Download URL	tako_sdk-2.2.20.tar.gz
Size	163.0 kB
Tags	Source
SHA-256 checksum How to use checksums	`dd7a71f03b686788ab28f4ce638d07e0875be900cdfc1225bb33a894c1eb4eed`
BLAKE2b-256 checksum How to use checksums	`84e4533b9b6a9e635f810e62c9ca6c978d59bd636f6a5518a04f6424fca95c8c`
Upload date	Aug 25, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`uv/0.10.2 {"installer":{"name":"uv","version":"0.10.2","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}`

Release files / tako_sdk-2.2.20-py3-none-any.whl

Download URL	tako_sdk-2.2.20-py3-none-any.whl
Size	380.1 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`04533bd721418c1ce660614db6bd20436adf3dca8b3ebd2e984b02482d7ce8c5`
BLAKE2b-256 checksum How to use checksums	`a528bcf4655c99088ac3c1987944fe5ff0ce5e959a4c80e27ca651d5991e0b7a`
Upload date	Aug 25, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`uv/0.10.2 {"installer":{"name":"uv","version":"0.10.2","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}`