tex-sdk 0.3.0

Tex Python SDK for IntegrationBackend

Tex SDK (Python)

Tex is a lightweight Python client for Tex’s IntegrationBackend HTTP API.

This README is intentionally implementation-grounded: it documents exactly what the SDK does today based on the code in:

• tex/__init__.py (public exports)
• tex/config.py (endpoint paths + config)
• tex/client.py (client behavior, auth, retries, request/response handling)
• tex/errors.py (error types)

If you are reading this inside the repo, those files are the source of truth.

What you get

• One client class: Tex
• Three auth modes (API key, org+user login, or direct access token)
• Minimal, predictable return values: most methods return JSON as dict
• Friendly NLQ output via AskResult
• Typed package (tex/py.typed) for mypy/pyright

Install

pip install tex-sdk

Optional: enable HTTP/2 (requires h2):

pip install "tex-sdk[http2]"

Python requirement (from pyproject.toml): Python 3.9+

Quick start

Important: the SDK’s base_url must point at IntegrationBackend (the FastAPI gateway), not HelixDB directly.

1) API key auth (recommended)

This is the simplest and most “production-like” flow.

from tex import Tex

tx = Tex("http://localhost:8000", api_key="sk_live_...")

tx.store_memory(
    "Hello from Tex SDK.",
    type="document",
    metadata={"source": "quickstart"},
)

answer = tx.ask("What did I just store?")
print(answer.text)

2) Org + user login (dev / testing)

If your IntegrationBackend allows /auth/login for a given org+user, you can use:

from tex import Tex

tx = Tex(
    "http://localhost:8000",
    org_id="org_123",
    user_id="user_456",
    session_id="s1",  # optional
)

tx.store_memory("I like espresso.", type="document")
print(tx.ask("What do I like?").text)

3) Direct access token (bring-your-own JWT)

If you already obtained an access token (e.g., out-of-band), pass it directly:

from tex import Tex

tx = Tex("http://localhost:8000", access_token="eyJ...")
print(tx.whoami())

The public API surface

The package exports (see tex/__init__.py):

• Tex
• AskResult
• TexConfig, TexEndpoints
• TexError, TexAuthError, TexHTTPError

Concepts

Multi-tenant scope

IntegrationBackend is multi-tenant. Requests typically belong to a tenant:

• org_id
• user_id
• session_id (optional)

In the SDK, “scope” is included in ingestion payloads via Tex._scope_payload(). Internally:

• The backend ultimately trusts the JWT for tenant claims.
• The SDK still sends scope because the request models expect it.
• When using API keys, the SDK attempts to discover tenant claims by calling GET /auth/verify.

If tenant discovery fails, the SDK may send placeholders ("_") for org_id/user_id; the backend is expected to ignore these and use the JWT tenant instead.

Correlation IDs

Each request includes a unique X-Correlation-ID header generated per request (see tex/client.py).

• This is useful for tracing requests through your logs.
• When available, the SDK also exposes this as request_id on raised exceptions.

Authentication: exact behavior

This section mirrors Tex._ensure_auth() and Tex._refresh() in tex/client.py.

Auth modes (in priority order)

When the SDK needs auth, it chooses the first available:

1. If access_token is already present in memory → use it.
2. If TexConfig.access_token is provided → use it.
3. Else if api_key is provided:
   • POST /auth/token-exchange with { "api_key": "..." } → get access_token (+ optional refresh_token)
   • GET /auth/verify to discover tenant claims
   • If you also explicitly provided user_id, the SDK then calls POST /auth/login to mint a user-scoped token.
4. Else if org_id + user_id is provided:
   • POST /auth/login with { org_id, user_id, session_id }
5. Otherwise → raise TexAuthError (“No auth configured...”).

Refresh + retry

All authenticated requests go through _request().

• If a request returns HTTP 401:
  • The SDK attempts _refresh() and retries the request once.
• Refresh rules:
  • If refresh_token exists: POST /auth/refresh.
  • Else if api_key exists: re-exchange the API key (and, if user_id was set, re-login).
  • Otherwise: raise TexAuthError.

Configuration

TexConfig

You can construct the client with either:

• A base_url string: Tex("http://localhost:8000", api_key=...)
• A TexConfig object: Tex(TexConfig(base_url=..., api_key=...))

Fields (see tex/config.py):

• base_url: IntegrationBackend URL, e.g. http://localhost:8000
• Auth fields: api_key, org_id, user_id, session_id, access_token, refresh_token
• Transport: timeout_s (default 15s), http2 (default True)
• endpoints: a TexEndpoints instance

TexEndpoints

TexEndpoints contains the path strings the SDK calls (all relative to base_url).

Defaults (see tex/config.py):

• Auth:
  • auth_login: /auth/login
  • auth_refresh: /auth/refresh
  • auth_token_exchange: /auth/token-exchange
  • auth_verify: /auth/verify
• Ingestion:
  • ingestion_document: /ingestion/document
  • ingestion_episode: /ingestion/episode
  • ingestion_preference: /ingestion/preference
  • ingestion_status: /ingestion/status/{job_id}
  • ingestion_batch: /ingestion/batch
• DB:
  • db_query: /helixdb/query
  • db_schema: /helixdb/schema
• NLQ: nlq_execute: /nlq/execute
• Search: search: /search
• Memories CRUD: memories_list, memories_get, memories_delete, memories_update
• Episodes: episodes_list: /memories/episodes
• Users: user_profile: /users/profile

If your deployment uses different routes/prefixes, override:

from tex import Tex, TexConfig, TexEndpoints

endpoints = TexEndpoints(
    # example override
    db_query="/db/query",
    db_schema="/db/schema",
)

tx = Tex(TexConfig(base_url="https://api.example.com", api_key="sk_live_...", endpoints=endpoints))

API reference (method-by-method)

All examples assume:

from tex import Tex

tx = Tex("http://localhost:8000", api_key="sk_live_...")

Tex.store_memory(...)

Single ingestion entrypoint. Behavior depends on type.

Signature (from tex/client.py):

• content: str or list[dict] (depending on type)
• type: one of "document", "episode", "preference"
• format: for documents, default "text"
• metadata: optional dict
• options: optional dict (passed through)
• episode_id: only for episode ingestion

Document ingestion (type="document")

resp = tx.store_memory(
    "A short document to ingest.",
    type="document",
    format="text",
    metadata={"source": "docs"},
)
print(resp)

Notes:

• For type="document", content must be a string or the SDK raises ValueError.
• The request payload includes scope derived from your auth.

Episode ingestion (type="episode")

Episodes represent chat-like messages.

You can pass a single string (it becomes one {"role":"user","content":...} message):

resp = tx.store_memory(
    "Today I called Alice and discussed the plan.",
    type="episode",
)

Or pass a message list:

messages = [
    {"role": "user", "content": "Book a table for two."},
    {"role": "assistant", "content": "Which restaurant?"},
    {"role": "user", "content": "Somewhere near downtown."},
]

resp = tx.store_memory(messages, type="episode", episode_id="ep_001")

If you pass an invalid structure, the SDK raises ValueError.

Preference ingestion (type="preference")

Preference ingestion expects preferences in metadata["preferences"].

resp = tx.store_memory(
    "ignored",
    type="preference",
    metadata={
        "preferences": [
            {"key": "drink", "value": "espresso", "confidence": 0.9},
        ]
    },
)

Important:

• For type="preference", the SDK ignores content and requires a non-empty list at metadata["preferences"].
• The SDK sends { org_id, user_id, session_id, preferences } using discovered scope.

Tex.job(job_id)

Fetch background ingestion status.

status = tx.job("job_...")
print(status)

Internally calls GET /ingestion/status/{job_id}.

Tex.batch_store(documents)

Ingest multiple documents in one call.

Each document dict should contain:

• data (str)
• optional format, metadata, options

resp = tx.batch_store(
    [
        {"data": "doc 1", "metadata": {"source": "batch"}},
        {"data": "doc 2", "format": "text"},
    ]
)
print(resp)

Return shape depends on IntegrationBackend; the SDK returns the JSON response.

Tex.search(query, ...)

Fast semantic search.

resp = tx.search("espresso", top_k=5)
print(resp)

Optional parameters:

• min_score: float
• label: string label filter
• metadata_filter: dict

Tex.ask(question, ...) → AskResult

Natural language query execution.

ans = tx.ask("What did I store recently?")
print(ans.text)

Return type is AskResult:

• text: human-readable summary assembled from evidence/bindings/documents
• evidence: list of evidence strings
• entities: list of extracted/bound entity names
• documents: list of document identifiers
• raw: the full JSON response dict

Parameters (passed through to /nlq/execute):

• execute (default True)
• enable_pruning (default True)
• use_local_intelligence (default True)
• intent_match_options (optional dict)
• compact (default True)

Tex.query(query, params=None)

Execute a DB query via IntegrationBackend.

resp = tx.query(
    "MATCH (n) RETURN n LIMIT $limit",
    params={"limit": 5},
)
print(resp)

This calls POST /helixdb/query by default (see TexEndpoints.db_query).

Tex.schema()

Fetch the database schema.

schema = tx.schema()
print(schema)

This calls GET /helixdb/schema by default.

Tex.whoami()

Returns the tenant context as seen by IntegrationBackend (great for debugging auth/scope).

print(tx.whoami())

Internally, it ensures auth, then calls GET /auth/verify (unless already cached).

Memories CRUD

These map to /memories endpoints.

Tex.get_memory(memory_id)

mem = tx.get_memory("mem_...")
print(mem)

Tex.list_memories(type=None, limit=50, offset=0)

resp = tx.list_memories(limit=20, offset=0)
print(resp)

docs = tx.list_memories(type="document", limit=20)
print(docs)

Tex.update_memory(memory_id, content=None, metadata=None, options=None)

resp = tx.update_memory(
    "mem_...",
    content="updated content",
    metadata={"tag": "updated"},
)
print(resp)

Tex.delete_memory(memory_id)

resp = tx.delete_memory("mem_...")
print(resp)

Tex.delete_memories(user_id=None)

Bulk delete.

• If user_id is omitted, deletes the caller’s memories.
• If user_id is provided and differs from caller, IntegrationBackend may require elevated roles.

resp = tx.delete_memories()
print(resp)

Episodes

Tex.list_episodes(limit=50, offset=0, since=None)

resp = tx.list_episodes(limit=10)
print(resp)

User profile

Tex.get_profile(format="text")

Returns a synthesized profile derived from preferences and episodic memories.

profile = tx.get_profile(format="text")
print(profile)

Errors and exception handling

Errors are defined in tex/errors.py.

TexHTTPError

Raised for non-auth HTTP failures (status >= 400 excluding 401/403) and network/timeout errors.

Fields:

• message (str)
• status_code (optional int)
• request_id (optional str; pulled from X-Correlation-ID response header if present)
• response_text (optional str; truncated to 2000 chars)
• details (any; parsed JSON when possible)

TexAuthError

Subclass of TexHTTPError, raised for auth issues:

• “No auth configured”
• token exchange/login failures
• HTTP 401/403 responses

Typical pattern

from tex import Tex, TexAuthError, TexHTTPError

tx = Tex("http://localhost:8000", api_key="sk_live_...")

try:
    print(tx.whoami())
except TexAuthError as e:
    # Wrong key, missing roles, expired/invalid refresh, etc.
    print("auth failed", str(e), e.status_code)
except TexHTTPError as e:
    # Non-auth HTTP errors (422, 500, network issues)
    print("request failed", str(e), e.status_code)

Request/response handling details

This mirrors Tex._request() and _post_noauth().

• JSON responses are returned as Python dicts.
• If the response body is valid JSON but not a dict (e.g., a list), the SDK wraps it as { "data": <payload> }.
• If the response is not JSON, the SDK returns { "data": "<text>" }.
• For error responses, the SDK tries to extract detail or message from JSON. Otherwise: “Request failed”.

Special hint for HTTP 422:

• If the error mentions “IntentGraph validation” / “root variable must have a type”, the SDK appends a hint suggesting schema/planner issues.

Running the smoketest script (repo)

The repo includes a minimal end-to-end tester:

• tools/tex_sdk_smoketest.py

It is designed to work:

• When installed from PyPI (pip install tex-sdk)
• When run directly from this repo (it falls back to adding the repo root to sys.path)

Environment variables

• TEX_BASE_URL (default: http://localhost:8000)
• TEX_TIMEOUT_S (default: 30)

Auth (set exactly one of the following groups):

1. API key:

• TEX_API_KEY=sk_live_...

2. Org+user login:

• TEX_ORG_ID=...
• TEX_USER_ID=...
• TEX_SESSION_ID=... (optional)

3. Direct token:

• TEX_ACCESS_TOKEN=eyJ...

Run

$env:TEX_BASE_URL = "http://localhost:8000"
$env:TEX_API_KEY = "sk_live_..."

python tools/tex_sdk_smoketest.py

The script performs:

• GET /health (unauth quick check)
• whoami()
• store_memory(type="document")
• polls job(job_id) (if job_id returned)
• search(...)
• ask(...)

Migration notes (in-repo users)

Inside this repo, there is also an sdk/ package that re-exports everything from tex/ as a compatibility shim.

• New code should import from tex.
• Old code importing sdk will continue to work (see sdk/__init__.py).

Troubleshooting

“Connection refused” / health check fails

• Ensure IntegrationBackend is running (default: http://localhost:8000/health).
• The SDK does not start services; it only calls HTTP endpoints.

HTTP 401 / 403

• Verify you’re using the correct auth mode.
• For API key auth: ensure the key is valid and belongs to the tenant you expect.
• For org+user login: ensure /auth/login is enabled for that org/user.

HTTP 422 (validation errors)

• Usually means your request payload doesn’t match backend expectations.
• For NLQ/planner-related validation errors, confirm the DB schema is available and the planner can fetch it.

HTTP/2 import error (h2)

If you see an ImportError mentioning h2, install:

pip install "tex-sdk[http2]"

Or disable HTTP/2:

from tex import Tex

tx = Tex("http://localhost:8000", api_key="sk_live_...", http2=False)