Talkory Python SDK

Installation

The SDK targets Python 3.9+, is built on httpx and Pydantic v2, and ships fully typed with a py.typed marker so your editor and type checker understand every response shape.

pip install talkory

Quickstart

Every response is a typed object: result.consensus is the synthesized answer, result.common_answer is the content every model agreed on, and result.confidence is a 0-100 score.

from talkory import Talkory

client = Talkory()  # reads TALKORY_API_KEY from environment
result = client.queries.run("Best commuter motorcycles under $5000?")

print(result.consensus)      # synthesized answer
print(result.common_answer)  # agreed-upon content
print(result.confidence)     # 0-100 confidence
print(result.charged)        # USD cost

Authentication

Create a key at app.talkory.ai → Settings → API Keys, then either set it as an environment variable or pass it explicitly:

export TALKORY_API_KEY="tk_live_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

from talkory import Talkory

# reads TALKORY_API_KEY automatically
client = Talkory()

# or pass a key explicitly
client = Talkory(api_key="tk_live_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx")

Sync & async clients

Use Talkory for scripts and simple sync code, or AsyncTalkory as an async context manager inside asyncio applications.

import asyncio
from talkory import AsyncTalkory

async def main():
    async with AsyncTalkory() as client:
        result = await client.queries.run("Best NAS for home use?")
        print(result.consensus)

asyncio.run(main())

Streaming

client.queries.stream() yields typed events as each model responds, ending with a done event carrying the consensus.

for event in client.queries.stream("Compare Python web frameworks in 2026"):
    if event.type == "model_chunk":
        print(event.chunk, end="", flush=True)
    elif event.type == "done":
        print("\n\nConsensus:", event.consensus)

Key features

1. Model selection. Choose up to 10 models per query, or omit the field to run everything on your plan.
2. Recursive review. Optional self-review round where models critique and improve their own answers before consensus.
3. Localization. Pass a country code to localise answers.
4. Deferred consensus. Run queries cheaply with consensus=False, then compute the consensus later via queries.consensus().
5. Fire-and-forget. Create a query, then poll or wait for the result asynchronously.
6. Wallet safety. The SDK never auto-retries paid POST requests, so a network blip can't double-charge your wallet.

Client reference

result = client.queries.run(
    "Best commuter motorcycles under $5000?",
    models=["gpt", "claude", "gemini"],
    recursive=False,
)
print(result.consensus, result.common_answer, result.confidence, result.charged)

for event in client.queries.stream("Compare Python web frameworks in 2026"):
    if event.type == "model_chunk":
        print(event.chunk, end="", flush=True)
    elif event.type == "model_result":
        print(f"\n{event.model_id} scored {event.score}")
    elif event.type == "done":
        print("\n\nConsensus:", event.consensus)

query = client.queries.create("Best commuter bikes under $5000?")
print(query.id)  # save this - the run continues even if you disconnect

result = client.queries.wait(query.id)  # blocks until status leaves "running"
print(result.consensus)

result = client.queries.run("Best NAS for home use?", consensus=False)
consensus = client.queries.consensus(result.id)
print(consensus.consensus, consensus.confidence)

query = client.queries.retrieve("7f9e4c2a-1b8f-4a21-9d3c-e77aa53b6c10")
print(query.status)

for q in client.queries.list(limit=20):
    print(q.id, q.prompt, q.status)

print(client.models.list())
print(client.health.check().status)
print(client.wallet.get().balance)
print(client.usage.get().requests.total)

Error handling

The SDK raises typed exceptions carrying code, status, message, and request_id, so you can branch on error type or log the request id for support.

from talkory import Talkory, RateLimitError, InsufficientBalanceError, TalkoryError

client = Talkory()
try:
    result = client.queries.run("Best commuter motorcycles under $5000?")
except RateLimitError as e:
    print("slow down:", e.request_id)
except InsufficientBalanceError as e:
    print("top up your wallet:", e.message)
except TalkoryError as e:
    print(e.code, e.status, e.message, e.request_id)

Security

FAQ

Need a key? Head to your API settings. Prefer REST? See the API documentation. Package details, release history and older versions live on PyPI ↗, and the source is on GitHub ↗.