Skip to content
BranchesTags

Repository files navigation

tensicai — the official Python SDK for tensic.ai

PyPI version Python versions CI License

Talk to a tensic.ai server from your own product: chat with RAG, agent and router projects, stream answers, keep conversations, manage the knowledge base, mint scoped API keys, and use the OpenAI-compatible endpoints — from a sync (Tensic) or async (AsyncTensic) client with full type hints and two runtime dependencies (httpx and anyio, the async layer httpx.AsyncClient itself runs on).

Install

pip install tensicai

Requires Python 3.10+ and a tensic.ai server (self-hosted, or a hosted instance from tensic.ai). Create an API key in the tensic.ai UI (Users → API keys) and export it:

export TENSIC_URL="https://tensic.example.com"   # your server
export TENSIC_API_KEY="sk-..."                    # a user or project-scoped key

60-second example

from tensicai import Tensic

client = Tensic()  # reads TENSIC_URL and TENSIC_API_KEY from the environment

response = client.chat.send(project_id=1, question="What is our refund policy?")
print(response.answer)
for source in response.sources:
    print(f"- {source.source} (score {source.score:.2f})")
tokens = response.tokens.total if response.tokens else 0
print(f"{tokens} tokens, conversation id {response.id}")

chat.send() runs one turn against a project. Every response is a lenient dataclass (ChatResponse) — unknown fields never break parsing (response.extra, response.raw). Pass chat_id= to continue an existing conversation, or use Conversations.

Streaming

Streaming yields typed events as the server produces them. stream.text() gives just the answer deltas; iterating the stream gives every event (plans, steps, tool calls, warnings, the final response).

from tensicai import Tensic
from tensicai.types import TextEvent, ToolCallStartedEvent

client = Tensic()

with client.chat.stream(project_id=1, question="Summarise this week's tickets") as stream:
    for event in stream:
        if isinstance(event, TextEvent):
            print(event.text, end="", flush=True)
        elif isinstance(event, ToolCallStartedEvent):
            print(f"\n[calling {event.tool}]")

print()
final = stream.response  # the authoritative ChatResponse, or None if the stream broke early
print("answer:", stream.answer)
print("chat id:", stream.chat_id, "| last event id:", stream.last_event_id)

Or just the text:

from tensicai import Tensic

client = Tensic()
for delta in client.chat.stream(project_id=1, question="Tell me a story").text():
    print(delta, end="")

Details worth knowing:

  • Event classes live in tensicai.types: TextEvent, PlanEvent, StepStartEvent, StepDoneEvent, ToolCallStartedEvent, ToolCallCompletedEvent, WarningEvent, ErrorEvent, StoppedEvent, ResponseEvent, UnknownEvent. All carry .type, .raw and .event_id.
  • If the server emits an error frame, the stream yields the ErrorEvent and then raises ChatStreamError (.code, .message, .chat_id, .last_event_id). Pass raise_on_error=False to inspect stream.error yourself instead.
  • Resume an interrupted turn by re-sending the same request with last_event_id=stream.last_event_id and the same chat_id; the server replays what you missed and tails the live run.
  • Cancel an in-flight turn from anywhere with client.chat.stop(project_id, chat_id).
  • Events are buffered, so stream.answer, stream.response and stream.events() keep working after iteration ends.

Conversations

A Conversation remembers the project and chat id for you (a uuid4 hex id is generated when you do not pass one) and collects every ChatResponse in .history:

from tensicai import Tensic

client = Tensic()
conversation = client.chat.conversation(project_id=1)  # or chat_id="support-42"

first = conversation.send("Which plans include SSO?")
follow_up = conversation.send("And how much does the cheapest one cost?")

print(conversation.id, len(conversation.history), "turns")
print(follow_up.answer)

for delta in conversation.stream("Thanks — one-line summary?").text():
    print(delta, end="")
conversation.stop()  # cancels an in-flight turn, no-op otherwise

RAG knowledge

RAG projects own a knowledge base. Ingest text, URLs or files, search it, and manage sources:

from tensicai import Tensic

client = Tensic()
project_id = 1

result = client.knowledge.ingest_text(
    project_id,
    "Refunds are accepted within 30 days of purchase.",
    source="policies/refunds.md",
    keywords=["refund", "policy"],
)
print(f"{result.chunks} chunks from {result.documents} document(s)")

client.knowledge.ingest_url(project_id, "https://example.com/handbook")
client.knowledge.ingest_file(project_id, "handbook.pdf", method="auto", chunks=512)

for hit in client.knowledge.search(project_id, "refund window", k=5, score=0.2):
    print(hit.source, hit.score, hit.id)

print(client.knowledge.list_sources(project_id))
chunks = client.knowledge.get_source(project_id, "policies/refunds.md")
print(len(chunks.ids), "chunks stored")

client.knowledge.delete_source(project_id, "policies/refunds.md")

Large batches go through knowledge.ingest_bulk(project_id, [paths...]) which queues jobs server-side (knowledge.bulk_jobs() / cancel_bulk_job()), and knowledge.reembed() / reembed_status() rebuild the vectors after changing the embedding model. ingest_file accepts a path, bytes, an open binary file, or a (filename, content[, content_type]) tuple. Source names are sent to the server base64-encoded exactly like the web UI does, so any string is a valid source name.

Attachments and vision

Send images to vision-capable models and files to agents. Files can be sent inline (base64) or uploaded first, which is better for anything larger than a few hundred KB:

from tensicai import Tensic

client = Tensic()

# Vision: an http(s)/data URL, a base64 string, raw bytes or a pathlib.Path.
reply = client.chat.send(1, "What is in this picture?", image="https://example.com/cat.jpg")
print(reply.answer)

# Inline files (max 10 per turn): paths, bytes, open files, tuples or ready-made dicts.
reply = client.chat.send(1, "Summarise the attached report", files=["q3-report.pdf"])

# Upload once, reference many times (needs object storage configured on the server).
attachment = client.chat.upload_attachment(1, "q3-report.pdf")
reply = client.chat.send(1, "List the action items", files=[attachment])
print(attachment.upload_id, attachment.expires_at)

# Files the agent wrote to /artifacts/ come back with a 24h asset token; fetch the bytes:
for artifact in reply.artifacts:
    if artifact.token:
        data = client.chat.get_asset(1, artifact.token, download=True)
        print(artifact.name, artifact.mime_type, len(data), "bytes")

# reply.image (vision projects) is the model's image as a base64 data URL, not a token.
if reply.image:
    print(reply.image[:30], "...")

image= is never a filesystem path. A str is the base64 payload itself or an http(s):// / data: URL (the server fetches an http(s) one for you); raw bytes and a pathlib.Path are read and base64-encoded by the SDK. So a local picture is image=Path("cat.png") or image=open("cat.png", "rb").read() — and passing the filename as a str raises ValueError before any request, pointing you at Path(...), rather than silently sending the model a filename it can never see. files= is the opposite: there a str is a path.

client.image.generate(...) takes the same values for its image-to-image image=, minus the http(s) URL: that endpoint base64-decodes whatever it is given instead of fetching, so pass base64, a data: URL, bytes or a Path.

Projects and API keys

from tensicai import Tensic

client = Tensic()

# Browse (auto-paging) and look up projects
for project in client.projects.iter():
    print(project.id, project.name, project.type, project.llm)
support = client.projects.find("support-bot")

# Create a RAG project and tune it (PATCH then GET — you always get the full object back)
project = client.projects.create(
    "support-bot",
    type="rag",
    team_id=1,
    llm="gpt-4o",
    embeddings="text-embedding-3-small",
    human_name="Support bot",
)
project_id = project.id
assert project_id is not None  # models are lenient, so ids are Optional to the type checker
project = client.projects.update(project_id, system="You are a concise support assistant.", k=6)

# Mint a read-only API key scoped to this project for your integration
key = client.users.api_keys.create(
    "integration-user",
    team_id=1,
    description="website chat",
    allowed_projects=[project_id],
    read_only=True,
)
print(key.api_key)  # shown once — store it now

Sub-resources hang off client.projects and take the project id first: prompts, secrets, routines, widgets, logs, conversations, analytics, evals, memory, comments, guards, custom_tools, webhooks, integrations. For example client.projects.widgets.create(project.id, name="Site chat", allowed_domains=["example.com"]) returns the widget key (current servers return the live key from every widget read too, so treat those payloads as secret-bearing), and client.projects.logs.list(project.id, has_error=True) lists failed turns.

projects.update(options=...) replaces the whole options blob server-side (the server fills defaults for keys you omit and only preserves sensitive ones), so read, modify, write:

from tensicai import Tensic

client = Tensic()
project = client.projects.get(1)
options = dict(project.options)
options["max_iterations"] = 8
project = client.projects.update(1, options=options)
print(project.options)

The same read-modify-write applies to every other option blob the API exposes: teams.update(options=/branding=), users.update(options=), llms.update(options=), embeddings.update(options=) and the image-generator / speech-to-text registries all replace what they are given rather than merging it key by key (masked "********" credentials are the one thing carried forward). classifiers.update(options=) is the single exception — that endpoint does merge.

Team membership and grants are addressed by name: teams.create(...) / teams.update(...) take usernames, project names and model names in users=, admins=, projects=, llms=, embeddings=; users.update(username, projects=[...]) likewise takes project names.

Async

AsyncTensic mirrors Tensic method for method; streaming methods are awaited and return an AsyncChatStream:

import asyncio

from tensicai import AsyncTensic


async def main() -> None:
    async with AsyncTensic() as client:
        me = await client.whoami()
        print("hello,", me.username)

        reply = await client.chat.send(project_id=1, question="Ping?")
        print(reply.answer)

        stream = await client.chat.stream(project_id=1, question="Stream me")
        async for delta in stream.text():
            print(delta, end="")

        async for project in client.projects.iter():
            print(project.name)


asyncio.run(main())

Errors and retries

Every HTTP failure raises a subclass of tensicai.APIStatusError (itself a TensicError) carrying status_code, the server's stable code, message, fields (validation details), request_id, retry_after, rate_limit, body, headers and the raw response.

Exception Status Typical code
BadRequestError 400 invalid_request
AuthenticationError 401 unauthenticated
BudgetExceededError 402 budget_exceeded
PermissionDeniedError 403 forbidden (e.g. read-only key)
NotFoundError 404 not_found
ConflictError 409 conflict
PayloadTooLargeError 413 payload_too_large
UnsupportedMediaTypeError 415 unsupported_media_type
ValidationError 422 validation_error (see .fields)
RateLimitError 429 rate_limited, quota_exceeded
InternalServerError 5xx internal
BadGatewayError 502 bad_gateway
ServiceUnavailableError 503 service_unavailable
APIConnectionError — network failure
APITimeoutError — timeout (subclass of the above)
TwoFactorRequiredError — Tensic.login() needs a TOTP code
ChatStreamError — error frame mid-stream
UnsupportedOperationError — the SDK cannot do this locally 
from tensicai import NotFoundError, RateLimitError, Tensic, ValidationError

client = Tensic()
try:
    client.chat.send(project_id=999, question="hi")
except NotFoundError as err:
    print(err.status_code, err.code, err.message, err.request_id)
except ValidationError as err:
    print(err.fields)  # {"body.question": "field required", ...}
except RateLimitError as err:
    print("retry in", err.retry_after, "s;", err.rate_limit)

Retries: transient failures (408, 429, 502, 503, 504 and connection errors) are retried up to max_retries times (default 2) with exponential back-off and jitter, honouring Retry-After. Only idempotent requests are retried — GET/PUT/DELETE, read-like POSTs such as search and classify, and POSTs carrying an Idempotency-Key. chat.send() generates one automatically (pass idempotency_key= to control it), so a retried chat turn is never executed twice; a 409 "still in flight" reply for that key is retried too. Streaming requests are never retried once the body has started. Disable retries with Tensic(max_retries=0) or per call site with client.with_options(max_retries=0).

Configuration

Setting Constructor argument Environment variable
Server URL Tensic(base_url=...) TENSIC_URL (alias TENSIC_BASE_URL); default http://localhost:9000
API key Tensic(api_key=...) TENSIC_API_KEY
Timeout timeout=60.0 or an httpx.Timeout —
Retries max_retries=2 —
Extra headers default_headers={...} —
TLS verification verify=True / CA bundle path / SSLContext —
Debug logging — TENSIC_LOG=debug (logger tensicai, no secrets) 
import httpx

from tensicai import Tensic

# Explicit configuration
client = Tensic(
    base_url="https://tensic.example.com",
    api_key="sk-...",
    timeout=httpx.Timeout(120.0, connect=5.0),
    max_retries=3,
    default_headers={"X-Team": "growth"},
)

# Per-call overrides without rebuilding the connection pool
patient = client.with_options(timeout=600.0, max_retries=0)
patient.knowledge.ingest_file(1, "big-manual.pdf")

# Long-running methods also accept timeout= directly
client.chat.send(1, "Deep research, please", timeout=300.0)

# Bring your own httpx client for proxies, custom transports or client certificates
http = httpx.Client(proxy="http://proxy.internal:3128", verify="/etc/ssl/corp-ca.pem")
client = Tensic("https://tensic.example.com", "sk-...", http_client=http)

# Clients are context managers; close() releases the pool
with Tensic() as scoped:
    print(scoped.version().version)

Username/password login is available too (it keeps the server's session cookie on the client):

from tensicai import Tensic, TwoFactorRequiredError

try:
    client = Tensic.login("alice", "s3cret", base_url="https://tensic.example.com")
except TwoFactorRequiredError:
    client = Tensic.login("alice", "s3cret", base_url="https://tensic.example.com", totp_code="123456")
print(client.whoami().username)
client.logout()

Instances with LDAP authentication enabled keep the credentials in the directory, so those accounts have no local password for HTTP Basic to verify and login() cannot sign them in. Use ldap_login(), which posts the credentials to /ldap instead:

from tensicai import Tensic

client = Tensic.ldap_login("alice", "s3cret", base_url="https://tensic.example.com")
print(client.whoami().username)
client.logout()

It yields the same tensic_token cookie session as login() (and AsyncTensic.ldap_login() is the async twin), with no TOTP step on this route. The account the session belongs to is named after the directory's mail attribute, so whoami().username is usually the address rather than the login name you passed. LDAP being disabled server-side, an empty password, a user missing from the directory and a failed bind all come back as BadRequestError.

Streaming requests use httpx.Timeout(None, connect=10.0) (no read timeout) unless you pass timeout=. The default timeout for everything else is 60 s with a 10 s connect timeout.

OpenAI-compatible direct access

The server exposes the OpenAI API surface under /v1 — no project needed, model is the tensic LLM name. Use the SDK:

from tensicai import Tensic

client = Tensic()

completion = client.direct.chat_completions(
    "gpt-4o",
    [{"role": "user", "content": "Say hello in Portuguese"}],
    temperature=0.2,
)
print(completion.content)

stream = client.direct.chat_completions("gpt-4o", [{"role": "user", "content": "Count to 5"}], stream=True)
for chunk in stream:
    print(chunk.content or "", end="")
print("\n", stream.content)  # the joined text

vectors = client.direct.embeddings("text-embedding-3-small", ["hello", "world"]).vectors
print(len(vectors), "vectors of", len(vectors[0]), "dimensions")

print([model.id for model in client.direct.models()])
image = client.direct.images_generate("a lighthouse at dawn", model="dall-e-3", size="1024x1024")
transcript = client.direct.audio_transcribe("meeting.mp3", model="whisper-1", language="en")

project_id= on chat_completions() / models() uses the project-governed variants (/projects/{id}/v1/...), which apply the project's guard and budget. raise_on_error=False on a streamed completion exposes stream.error instead of raising.

Or point the official openai package at the server:

from openai import OpenAI

from tensicai import Tensic

tensic = Tensic()
openai = OpenAI(base_url=tensic.openai_base_url, api_key=tensic.api_key)
print(openai.chat.completions.create(
    model="gpt-4o", messages=[{"role": "user", "content": "Hello!"}]
).choices[0].message.content)

Widget client

Published chat widgets authenticate with a widget key (wk_...) instead of a user key — the same thing the embeddable JavaScript does. WidgetClient reads TENSIC_URL and TENSIC_WIDGET_KEY when arguments are omitted.

from tensicai import WidgetClient

widget = WidgetClient("https://tensic.example.com", "wk_...")
print(widget.config().title)

reply = widget.chat("Do you ship to Portugal?")
print(reply.answer)
reply = widget.chat("How long does it take?", chat_id=reply.id)  # continue the conversation

for delta in widget.stream("And the cost?", chat_id=reply.id).text():
    print(delta, end="")

AsyncWidgetClient is the async twin. Pass context_token= to send a signed X-Widget-Context token.

More namespaces

Everything else the server offers is one attribute away; every method has a docstring with the endpoint it calls.

Namespace What it covers
client.teams list/iter/get/create/update/delete, add/remove users, admins and projects, grant/revoke models, analytics, transactions, branding, member budgets, invites
client.users CRUD, team budgets, TOTP setup/enable/disable/status, client.users.api_keys
client.llms, client.embeddings model registry CRUD (+ llms.test)
client.tools classify(sequence, labels) zero-shot classification, agent/MCP tool discovery, Ollama and OpenAI-compatible probes
client.statistics platform summary, daily tokens, top LLMs/projects, per-user usage
client.templates list/get/update/delete and instantiate(template_id, name, team_id=...)
client.image, client.audio generators/STT models available to you, generate(...), transcribe(...)
client.settings admin settings (update(values=None, **fields)), infrastructure, health, audit, cron logs, test_* connectivity checks
client.classifiers, client.image_generators, client.speech_to_text registry CRUD
client.invitations, client.examples, client.news invitations, example projects, platform news
client.search(query), client.whoami(), client.version(), client.health() / client.health(probe="live") global search, current user, server version, readiness / liveness probe (GET /health/ready, /health/live)

Anything not wrapped is reachable through the escape hatches client.get(path, params=...), client.post(path, json=...), client.patch, client.put, client.delete, which return the decoded body and raise the same typed errors.

Writing your own typed wrapper? The aliases the SDK's own signatures use are public too: from tensicai import NOT_GIVEN, NotGiven, FileInput, AttachmentInput, Timeout and from tensicai.types import ChatResponse, Privacy, Splitter, TemplateVisibility.

Not wrapped

A few routes are deliberately left to the escape hatch, because a typed method would not help anyone:

Endpoint Why
GET /oauth/{provider}/login, GET /oauth/{provider}/callback a browser redirect dance — the SDK never follows redirects, and the callback is called by the identity provider, not by you
GET/POST /webhooks/whatsapp inbound: Meta calls your server on it (verification handshake, then message delivery)
POST /mcp MCP JSON-RPC, which an MCP client (Claude, an IDE, the mcp package) speaks directly — point it at the URL and let it handle the protocol
POST /auth/support-login an operator-only exchange of a manager-signed, ~90-second, single-use grant for an admin session; not a customer integration path

If you do need one of them, the escape hatch is one line — same auth, same decoded body, same typed errors:

from tensicai import Tensic

client = Tensic()
tools = client.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
print(tools)

Project-scoped webhooks — the outbound ones your project calls — are wrapped: client.projects.webhooks.

Server compatibility

tensicai 0.1.x targets tensic.ai server 1.6 or newer. Models are lenient: fields added by newer servers land in .extra, and missing ones read as None, so a newer server never breaks an older SDK. Check what you are talking to with client.version().

Development

git clone https://github.com/tensicai/python-sdk && cd python-sdk
make install          # uv sync --extra dev
make check            # ruff check + ruff format --check + mypy src examples + pytest
make test-all         # pytest on Python 3.10 through 3.14 (uv fetches interpreters)

Tests never touch the network (they run against an in-memory httpx.MockTransport). Live tests in tests/integration/ run only when TENSIC_URL and TENSIC_API_KEY are set. See CONTRIBUTING.md for the full workflow and docs/INTERNALS.md for how the SDK is put together.

Releasing

  1. Bump __version__ in src/tensicai/_version.py and add the entry to CHANGELOG.md.
  2. Commit, then tag and push: git tag vX.Y.Z && git push origin main vX.Y.Z.
  3. The publish.yml workflow runs ruff, mypy and the full test suite, builds the sdist and wheel, checks that the tag matches the version, and publishes to PyPI through Trusted Publishing (no API token involved). A red suite fails the release before anything is uploaded.

One-time PyPI setup: add a pending Trusted Publisher on https://pypi.org/manage/account/publishing/ for project tensicai, owner tensicai, repository python-sdk, workflow publish.yml, environment pypi.

To rehearse a release, run the Publish workflow manually (Actions → Publish → Run workflow) with the Publish to TestPyPI input enabled: manual runs build and verify the package and upload it to TestPyPI only — production PyPI is reached exclusively by pushing a vX.Y.Z tag that matches __version__. For that rehearsal, also register a pending Trusted Publisher on https://test.pypi.org/manage/account/publishing/ with the same owner, repository and workflow, but environment testpypi.

License

Apache-2.0 — see LICENSE.

About

No description, website, or topics provided.

Resources

Readme
Apache-2.0 license

Contributing

Contributing
Activity
Custom properties

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors

Languages