Zeus Client Python Library

Python client library for Zeus AI data servers. Orchestrates LLM agents that call Zeus tools — catalog sync, auth, contracts, durable sessions, and the full agent loop — without any web UI.

2.4.1 — default import zeus_client is the journaled hexagonal ZeusRuntime tree. Temporary import zeus_client_v2 alias (deprecated). Claim remains candidate until human MATRIX. multi_agent is docs (CHECKLIST F). See docs/V2/, MIGRATION.md, CHANGELOG.md, MULTI_AGENT.md.

Claim (family honesty)

Pins file: sdk_bootstrap.pins.json (HOW_TO G0). Never invent production contract_hash / Hub stamps.

Family law and ship bar:

• HOW_TO_MAKE_A_CLIENT.md
• CHECKLIST.md
• MATRIX.md
• JAILBREAK_ATTEMPTS_SECURITY.md (floor-5 jailbreak catalog)
• COMPAT.md (engine pack triples — not invented here)

Note on ai_process_result: package default is false (cheap path). Use profile hub or ClientSettings(ai_process_result=True) for Hub Debug insight.

Note on session.semantic_cache: capability ships (CHECKLIST E2) with master enabled=false. Opt in via config; recall injects bag B semantic_memory; writes are explicit-only by default (rt.session.semantic_cache.write(...)). Zeus owns store/recall (POST /v2/agent_memory/*). Direct typeahead is never included. Requires a Zeus build that exposes agent_memory; older engines fail-open (no inject). MATRIX honesty: flag, not supported.

Install

pip install kotenai-zeus-client

Development install:

git clone https://github.com/koten-ai/zeus_client_python.git
cd zeus_client_python
pip install -e ".[dev]"

Optional OpenTelemetry Logs export:

pip install -e ".[otel]"

Product sinks stamp user=zeus_client (never Hub admin). Chat/turn ids are UUID v4 when the client mints them; Zeus mints X-Zeus-Req-Id unless a test pre-mints v4.

Version 0.3.0 — direct V2 verbs + run_<verb> naming

• run_verb / run_<verb> — direct V2 verb POSTs (no LLM) for every Zeus verb except pipeline (see docs/VERBS.md). Helpers: run_find, run_get, run_project, …; raw search body via run_search_verb / run_verb("search", …).
• run_search / run_search_from_config — no-LLM typeahead over V2 search (FTS) + optional N1QL hydrate + find→project (see docs/FAST_SUGGEST.md). Named after the primary Zeus verb (run_<verb>). Deprecated nicknames run_fast_suggest* were removed for GA (use run_search*).
• ai_process_result=false after Zeus tool data: no second LLM hop — prefer tool-arg summary, else a static thin line (UI shows Zeus rows).

Version 0.2.2 — fast suggest + thinner cheap path

• Typeahead + cheap path first land (API names later normalized in 0.3.0).

Version 0.2.1 — ai_process_result (Hub parity)

Client setting ai_process_result (default true) mirrors Zeus Hub Debug “AI on Zeus result”: after terminating Zeus tools, run a no-tools insight turn; set false for the cheap path (tables/UI without a second billable hop). See docs/BASE5_CLIENT.md.

Version 0.2.0 — base-5 control plane

Optional base-5 Client floor (object rules, Layer A policy table, base_id catalog load). See docs/BASE5_CLIENT.md. Legacy *_v2.json paths remain the default when base_id is omitted. Never invent production contract_hash values — Hub stamp only.

Agent flow

One run_agent call executes a single user turn through these phases:

flowchart LR
  A[Auth] --> B[Load catalog]
  B --> C[Contract + session]
  C --> D[LLM rounds]
  D --> E[Commit turn]
  E --> F[Audit]

1. Auth — resolve_zeus_auth mints or reuses Zeus session headers (per scope when configured).
2. Load catalog — rt.catalog.load(mode, base_id=...) resolves a stamped chat_request (scope dir → general → bundled; lineage files chat_request_{mode}_{base_id}.json). Brief / mini-schema inject is hash-excluded.
3. Contract + session — rt.catalog.contract.bind uses the published stamp (never invents a hash). Durable sessions create or rehydrate /v2/session; a mode switch starts a new session.
4. LLM rounds — multi-round tool loop: provider chat completion → Zeus dispatch (V2 verbs) → SecurityHooks + tool_trail inject. Force-return when the round budget is nearly exhausted.
5. Commit turn — multi-hop session-trace aggregate via post_session_trace (same rich body per tool req_id so every hop joins Detective; preferred hop last) plus a turn shard via continue_session_turn. Agent dispatches stamp X-Zeus-Mode (and optional X-Zeus-Trace: 1).
6. Audit — runtime contract checks appended to trace["notes"].

Pass zeus_session_id and zeus_round from the prior turn's session_meta to continue a durable session across questions.

Fast-tier search (typeahead, no LLM)

Google-like dropdowns should not call run_agent. Use the V2 data plane via run_search (named after the Zeus search verb):

from zeus_client import ZeusClient, run_search, SuggestOptions

async with ZeusClient():
    result = await run_search(
        "sushi",
        zeus_url="http://127.0.0.1:8080",
        bucket="yelp-data",
        scope="_default",
        collection="_default",
        zcfg={"auth_mode": "basic", "username": "…", "password": "…"},
        # Optional: hydrate FTS src_keys via Couchbase Query (USE KEYS)
        couchbase={
            "query_url": "http://127.0.0.1:8093",
            "username": "Administrator",
            "password": "password",
        },
        options=SuggestOptions(entity_type="Business", limit=8, fts_timeout_ms=2000),
    )
    for hit in result.hits:
        print(hit.name, hit.subtitle, hit.id)

What it calls:

Do not project / get FTS biz:… keys as graph node ids — they come back missing. Prefer N1QL hydrate or find→project.

Config helper: run_search_from_config(q, cfg) reads samples + zeus + optional couchbase. CLI sketch: examples/run_search.py. Module: zeus_client.zeus.suggest.

UI: debounce ~280ms, min query length 2; on Enter without a highlight, call run_agent for full NL search.

Direct V2 verbs (no LLM, no pipeline)

For BFF paths that need a single Zeus verb without the agent loop:

from zeus_client import ZeusClient, run_find, run_get, run_verb

async with ZeusClient():
    found = await run_find(
        {"entity_type": "Business", "where": {"name": "Pathmark"}, "limit": 5},
        zeus_url="http://127.0.0.1:8080",
        bucket="yelp-data",
        scope="_default",
        zcfg={"auth_mode": "basic", "username": "…", "password": "…"},
    )
    ids = [i.get("id") for i in (found.body or {}).get("items") or [] if isinstance(i, dict)]
    if ids:
        got = await run_get(
            {"ids": ids[:10], "include": ["body"]},
            zeus_url="http://127.0.0.1:8080",
            bucket="yelp-data",
            scope="_default",
            zeus_headers=…,  # or zcfg=
        )

Full list + auth: docs/VERBS.md. Module: zeus_client.zeus.verbs.

Quick start

Sync catalogs from your Zeus server once (recommended before the first agent run):

import asyncio
from zeus_client import (
    ZeusClient,
    load_config,
    resolve_llm_provider_config,
    resolve_zeus_config,
    run_agent,
    sync_chat_requests,
)


async def main():
    async with ZeusClient():
        cfg = await load_config()
        zcfg = resolve_zeus_config(cfg)

        # Pull stamped chat_request snapshots into ~/.config/zeus_client/chat_requests/
        result = await sync_chat_requests(cfg)
        print(f"Synced {len(result.synced)} catalog(s)")

        provider = resolve_llm_provider_config(cfg)
        sample = cfg["samples"][cfg["default_sample"]]

        answer, trace, turns, session_meta = await run_agent(
            zcfg["url"], zcfg,
            provider["base_url"], provider["api_key"], provider["models"][0],
            cfg.get("default_api_version", "v2"),
            cfg.get("default_mode", "analytics"),
            sample["bucket"], sample["scope"], sample["collection"],
            "How many breweries are in the dataset?",
            prior_turns=[],
        )
        print(answer)
        print("Session:", session_meta.get("session_id"))


asyncio.run(main())

See examples/minimal_agent.py for a shorter variant that skips sync (uses bundled catalogs).

Configuration

On first run, load_config() creates ~/.config/zeus_client/config.json from the bundled example template (src/data/config.example.json).

Override the config directory:

export ZEUS_CLIENT_CONFIG_DIR=/path/to/my/config

Zeus connection

Zeus settings live under a single top-level zeus object. Use resolve_zeus_config(cfg) in code to read it (applies runtime overrides such as ZEUS_URL).

"zeus": {
  "url": "http://localhost:8080",
  "auth_mode": "none",
  "username": "",
  "password": "",
  "bearer_token": "",
  "session_id": "",
  "notes": "",
  "scope_credentials": {
    "beer-sample/_default": { "username": "", "password": "" }
  },
  "scope_contracts": {
    "beer-sample/_default": {
      "contract_id": "",
      "contract_hash": ""
    }
  },
  "enable_durable_sessions": true
}

Runtime URL override (not written back to config.json):

export ZEUS_URL=http://localhost:8080

LLM provider

LLM settings live under a single top-level llm_provider object. Use resolve_llm_provider_config(cfg) in code to read it.

"llm_provider": {
  "label": "xAI Grok",
  "base_url": "https://api.x.ai/v1",
  "api_key": "",
  "models": [
    "grok-4-1-fast-non-reasoning",
    "grok-4-1-fast-reasoning",
    "grok-4",
    "grok-3-mini"
  ]
}

Any OpenAI-compatible endpoint works — point base_url and api_key at your provider and list the model IDs you want to use.

Catalog sync

chat_requests_sync in config controls background catalog pulls:

Synced files land under ~/.config/zeus_client/chat_requests/{bucket}__{scope}/ with a manifest.json fingerprint. load_chat_request prefers scope-specific synced files over bundled snapshots.

Contract hashes

In the normal flow you do not compute hashes on the client. Verify a standardized V2 chat_request with Zeus (Catalog → "Verify with Zeus"), save the stamped response, and sync or copy it into your catalog dir. The stamped contract.hash (or _hash) is what session create and drift checks use.

compute_contract_hash remains for offline diagnostics and legacy unstamped files only.

Lint catalog rules (open-surface conflicts)

Operators can inject unbounded rules into hash-excluded surfaces (guidance.*, especially guidance.injections.business_logic and guidance.optimal_paths). Over time those rules can contradict each other. The client ships a read-only linter:

Structured open rules (hard lint source of truth) look like:

{
  "id": "ban-find-beer",
  "effect": "forbid_tool",
  "tool": "find",
  "when": { "entity_type": "Beer" },
  "source": "plugin",
  "kind": "routing",
  "rule": "Do not use find for Beer."
}

from zeus_client import lint_chat_request, lint_catalog_assembled, hash_policy_summary

report = lint_chat_request(chat_req)
print(report.conflict_score, report.severity)  # score secondary; hard counts primary
print(report.hard_finding_count, report.soft_finding_count, report.open_vs_locked_count)
for f in report.findings:
    print(f.layer, f.severity, f.check_id, f.message)

# Assemble-time (cached by contract hash + open rules hash)
report = lint_catalog_assembled(chat_req)  # respects guidance.catalog_lint / env

print(hash_policy_summary())

CLI:

python scripts/lint_chat_request.py path/to/chat_request.json
python scripts/lint_chat_request.py path/to/chat_request.json --json
python scripts/lint_chat_request.py --policy
python scripts/lint_chat_request.py --schema          # structured open-rule schema
python scripts/lint_chat_request.py path/to/chat_request.json --fail-on hard  # CI

Config (per catalog guidance.catalog_lint, or env ZEUS_CATALOG_LINT_MODE / ZEUS_CATALOG_LINT_FAIL_ON):

mode: off | assemble | debug | ci
hard_conflicts / soft_nl / open_vs_locked: bool
fail_on: none | hard | high | medium | low   # chat never fails; CI uses hard

When guidance.debug is true (or mode is assemble/debug/ci), run_agent runs assemble-time lint once per catalog fingerprint (cached), attaches trace["catalog_lint"] in debug/ci or when hard findings exist, and never blocks the turn on soft findings.

See .grok/guides/CHAT_REQUEST_CONFLICT_LINT.md for V1 vs V2 detail, migration from free-text, and limitations of locked-side extraction.

Structured responses

Pass structured=True to run_agent to receive schema-filtered Zeus rows alongside the natural-language answer:

answer, trace, turns, meta, structured = await run_agent(
    zcfg["url"], zcfg,
    provider["base_url"], provider["api_key"], provider["models"][0],
    "v2", "analytics",
    sample["bucket"], sample["scope"], sample["collection"],
    "List the top Colorado breweries by name.",
    prior_turns=[],
    structured=True,
)

print(structured.answer)       # same as answer (summary from return verb)
print(structured.zeus_data)    # [{"id": "n_...", "name": "...", ...}, ...]
print(structured.decomposition)  # structured query understanding (if model emitted it)
print(structured.warnings)     # dropped unknown fields, missing schema, etc.

zeus_data rows are extracted from the last successful Zeus data tool call in trace["tool_calls"] (find, search, pipeline, project, etc.) and filtered to fields allowed by:

1. output_schema argument to run_agent (highest precedence), e.g. {"entity_type": "Hotel", "fields": ["id", "name", "rating"]}
2. guidance.injections.output_schema in the loaded chat_request (operator-defined, hash-excluded)
3. MINI-SCHEMA from the scope brief (get_mini_schema)

Reserved fields id, doc_key, and entity_type are always kept. FK join columns like brewery_id.name are kept when the prefix matches an entity_fk in MINI-SCHEMA. Unknown fields are dropped and listed in structured.warnings (also appended to trace["notes"] as [WARN] …).

You can also call extract_structured_response(answer, trace, chat_req) directly after a normal run_agent turn without re-running the agent.

Default structured=False preserves the original 4-tuple return value for backward compatibility.

Crawl / Walk / Run

Subclass AgentHooks and pass hooks=MyHooks() to run_agent to observe or steer the loop without forking core logic:

Public API

Key exports from import zeus_client:

Initialize HTTP manually if not using ZeusClient:

from zeus_client import init_http, close_http

await init_http()
# ... your code ...
await close_http()

Error handling

The client uses three strategies depending on severity:

Inspect trace after run_agent for session drift, auth re-mint notes, and the runtime contract audit block.

HTTP client lifecycle

Configuration (load_config)

Zeus authentication (resolve_zeus_auth)

401 self-heal: During tool dispatch, a 401 in basic mode triggers invalidate_zeus_session + forced re-login and one retry. Failure is noted in trace["notes"] as auth: re-mint after 401 failed (…).

Catalog loading

Catalog sync (sync_chat_requests)

Zeus tool dispatch (structured codes)

Returned as JSON when httpx fails (status 0). Zeus HTTP 4xx/5xx return the raw response text instead.

Durable sessions (structured codes and statuses)

Set zeus.enable_durable_sessions: false to skip session APIs when the server lacks session support.

LLM provider errors

Business logic validation (inject_business_logic, audit_rows_against_rules)

Runtime contract audit (trace["notes"])

After each turn, checks are appended as [PASS] / [FAIL] lines. Common failures:

If the audit itself throws, trace["notes"] contains runtime_audit_failed: … (non-fatal).

Demo Builder Kit

Build productized demos (NL search UI, agent loop, result cards, trace panel, Docker) using this library:

Minimal programmatic agent (no UI): examples/minimal_agent.py.

Build and publish

pip install build twine
python -m build
twine upload dist/*

Development quality gates

pip install -e ".[dev]"
make ci          # ruff + mypy + pytest/cov + build/twine

GitHub Actions (.github/workflows/):

Release (human): bump version clocks + CHANGELOG → merge → git tag vX.Y.Z && git push origin vX.Y.Z.
Do not self-award MATRIX / claim supported from CI green alone. pip-audit is non-blocking until ZCP-44.

License

BSD-3-Clause — see LICENSE.