talkory 0.0.5

Talkory: AI answer comparison and consensus platform. Official Python SDK.

talkory

Official Python SDK for Talkory.ai: ask once, hear from five LLMs (GPT, Claude, Gemini, Grok, Perplexity), get a consensus answer.

pip install talkory

Quickstart

Create an API key at app.talkory.ai (Settings → API Keys), then:

from talkory import Talkory

client = Talkory()  # reads TALKORY_API_KEY from the environment

result = client.queries.run("Best commuter motorcycles under $5000?")
print(result.consensus)       # synthesized consensus answer
print(result.common_answer)   # what the models agreed on
print(result.confidence)      # 0-100
print(result.charged)         # USD charged for this run

queries.run() blocks until the run finishes (can take a few minutes; the read timeout defaults to 600s per the API docs).

Streaming

Watch all five models answer live over Server-Sent Events:

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)

Events arrive as typed objects (query_created, model_stream_start, model_chunk, model_result, model_error, consensus_start, done, and more), so your IDE autocompletes the fields per event type. If the connection drops after the run starts, the SDK raises StreamInterruptedError carrying query_id; the run keeps going server-side, so recover with client.queries.retrieve(err.query_id) instead of re-running (which would charge twice).

Async

AsyncTalkory mirrors the entire client:

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)

        async for event in client.queries.stream("..."):
            if event.type == "model_chunk":
                print(event.chunk, end="")

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

asyncio.run(main())

Fire-and-forget + poll

run = client.queries.create("Compare Python web frameworks in 2026")
result = client.queries.wait(run.id)      # polls with gentle backoff

Choosing models

models = client.models.list()             # free
result = client.queries.run(
    "Is a heat pump worth it in Delhi?",
    models=["gpt", "claude"],             # subset (max 10)
    recursive=True,                       # round-2 self-review before consensus
    country="in",                         # localization
)

Free endpoints

client.health.check()                     # LLM provider health
client.wallet.get()                       # balance
client.usage.get(from_="2026-06-01")      # usage stats for this key
client.queries.retrieve(query_id)         # past result by id
for q in client.queries.list(limit=50):   # auto-paginating history
    print(q.id, q.prompt, q.charged)

Deferred consensus

result = client.queries.run("...", consensus=False)   # cheaper
later = client.queries.consensus(result.id)           # idempotent; cached re-calls charge $0

Errors

Every documented API error maps to a typed exception carrying code, status, message, and request_id (quote the request_id in support tickets):

from talkory import RateLimitError, InsufficientBalanceError, TalkoryError

try:
    result = client.queries.run("...")
except RateLimitError as e:
    time.sleep(e.retry_after or 30)
except InsufficientBalanceError:
    ...  # top up at app.talkory.ai
except TalkoryError as e:
    print(e.code, e.request_id)

Wallet safety

The SDK never auto-retries paid POST requests. If a connection drops after a run starts, the run still completes (and charges) server-side, and retrying blindly would double-spend your wallet. On timeout/disconnect, check client.queries.retrieve(id) first. Free idempotent GETs are retried automatically on 429/500/503, honoring Retry-After.

Security

• Key resolution: explicit api_key= argument → TALKORY_API_KEY env var. The SDK never reads or writes keys to disk.
• Keys are redacted everywhere they could leak: repr(), exception messages, logs show tk_live_***a1b2 at most.
• HTTPS is enforced; a plain-HTTP base_url is refused unless you explicitly pass allow_insecure=True (local testing only).
• Zero telemetry: the SDK makes no network calls other than the API requests you invoke. The only metadata sent is User-Agent: talkory-python/{version} python/{py_version}.
• Never embed keys in frontend/browser code; call Talkory from your backend.

See SECURITY.md for vulnerability reporting.

Requirements

Python 3.9+ · httpx · pydantic v2. Fully typed (py.typed).

Links

• API reference
• Dashboard / API keys
• Changelog

License

MIT