Python SDK

The import package is caveman_cloud. It uses only the standard library (urllib.request for HTTP, no requests, no httpx, dependencies = []) and requires Python 3.13 or newer. It wraps the gateway's /sdk/v1/* contract. It is not a compressor, tokenizer, or ranker; those decisions stay in the gateway and engine.

It mirrors the TypeScript SDK field-for-field — the snake_case names are the only difference. The two are kept in lockstep by a shared parity suite; a field that exists in one and not the other fails CI.

Install

The distribution name in pyproject.toml is caveman, but that PyPI name is not yet verified as ours, so there is no pip install caveman to rely on today. Install from source in editable mode and import caveman_cloud:

# from public/sdk/python (the package dir)
pip install -e .
# run the tests (Python >= 3.13)
pytest

from caveman_cloud import Cave, CompressResult, ToolSearchResult

The client

Cave(api_key, base_url, agent) takes three required fields — an API key, the gateway base URL, and an agent slug that labels every span. default_workflow, retention, and control_url (the control-plane host the jobs client targets) are optional.

from caveman_cloud import Cave

cave = Cave(
    api_key=os.environ["CAVE_API_KEY"],
    base_url="http://127.0.0.1:8787",
    agent="my-service",
    default_workflow="support-triage",  # optional; defaults to "unlabeled-workflow"
)

Tracing

cave.trace(workflow, tags) is a context manager yielding a Trace. Its tool(), model, artifacts, and checkpoint/expand helpers attribute everything to the trace's workflow. Tool spans are POSTed to /sdk/v1/events; telemetry failures are swallowed so they never break your agent.

with cave.trace("refund-flow", {"tier": "pro"}) as trace:
    order = trace.tool("lookup_order", {"read_only": True}, lambda: db_find(order_id))

    reply = trace.model["openai"].responses.create(
        {"model": "gpt-4.1", "input": f"Summarize order {order_id}"}
    )

Checkpoints

trace.checkpoint(messages, options) POSTs a conversation to /sdk/v1/checkpoints; the gateway persists it and returns a reversible source_ref. trace.expand(source_ref) is the other half — it GETs the stored {source_ref, version, messages, checkpoint} back. A checkpoint that cannot be expanded is a bug.

cp = trace.checkpoint(messages, {"reason": "pre-summarization"})
# ... later, or in a peer step ...
restored = trace.expand(cp["source_ref"])

Artifacts

trace.artifacts.page(value, options) pages a large payload out to /sdk/v1/artifacts and returns a compact [cave-artifact …] stub the model can expand on demand. strategy="verbatim" bypasses storage and returns the value unchanged. (trace.page_artifact is a backwards-compatible alias.)

stub = trace.artifacts.page(
    big_search_result,
    {"source": "vector-search", "content_type": "application/json", "strategy": "json-index"},
)

Tool search

cave.tools(catalog, *, strategy="all", initial_tool_count=8) returns a handle with .strategy, .initial, and .search(). strategy="all" puts the whole catalog in .initial; strategy="deferred" puts only the always_load tools plus the first initial_tool_count. Either way, .search() calls the gateway with the full catalog.

from caveman_cloud import CaveTool

catalog = [CaveTool(name="refund", description="Refund a charge", input_schema={...})]
tools = cave.tools(catalog, strategy="deferred")
# send tools.initial on the first turn

found: ToolSearchResult = tools.search("refund a charge", max_tools=5, ranker="bm25")
print(found.tools, found.saved_tokens, found.reduction_pct)

The ranker ("bm25" | "embeddings") is passed through to the gateway verbatim; the gateway honors "embeddings" only when it has an embedding provider wired. The SDK never computes similarity itself. cave.tool_search(tools, query, ...) is the flat variant for when you manage the catalog outside a handle.

Compression

cave.compress(payload, *, content_type=None) POSTs to /sdk/v1/compress and returns the Engine's report. The SDK delegates — it never reimplements a compressor, because that would fork behavior across surfaces.

result: CompressResult = cave.compress(big_json, content_type="json")
if result.recovery_handle:
    # dropped detail is recoverable byte-exact via the handle
    ...
print(result.ratio, result.basis)  # e.g. 0.93, "inferred"

Provider clients

cave.openai(), cave.anthropic(), cave.gemini(), and cave.vertex() return Provider objects proxied through the gateway. Each exposes a .raw(path, body) escape hatch for native provider paths the typed helpers don't cover. cave.bedrock(region) returns a plain dict (sdk_only=True) and makes no HTTP call.

openai = cave.openai(upstream_key=os.environ["OPENAI_API_KEY"])
out = openai.responses.create({"model": "gpt-4.1", "input": "hello"})

# vertex upstream_key is a Google OAuth2 access token
vertex = cave.vertex(upstream_key=gcloud_access_token())
native = vertex.raw(
    "/v1/projects/p/locations/us-central1/publishers/google/models/gemini-1.5-pro:generateContent",
    body,
)

You can pass a latency_class ("interactive" | "background" | "offline") to a responses.create call; anything other than "interactive" sets the x-cave-async header so the gateway can defer it.

The OTel exporter

cave.exporter(service_name=None) returns an OTelExporter bound to this client's gateway config. record_span() maps GenAI fields (provider, model, token counts, cost_usd) to gen_ai.* semantic-convention attributes; export() POSTs the buffered OTLP/JSON batch to /otlp/v1/traces and clears the buffer. No external OpenTelemetry wiring, no third-party deps — the payload is built by hand.

otel = cave.exporter(service_name="billing-agent")
root = otel.record_span(
    "chat.completion",
    provider="openai",
    model="gpt-4.1",
    input_tokens=1200,
    output_tokens=340,
    status="ok",
)
otel.record_span("tool.call", tool_name="lookup_order", parent_span_id=root.span_id)
otel.export()  # → {"ok": ..., "spans_accepted": ..., "spans_total": ...}

The retry-loop breaker

cave.retry_loop_breaker(threshold=3) returns a RetryLoopBreaker that interrupts a stuck agent re-issuing the same tool call. Call record(name, args) before each invocation (or wrap it with guard()); once the identical (name, args) signature repeats past the threshold it raises RetryLoopError. Any different call resets the streak.

from caveman_cloud import RetryLoopError

breaker = cave.retry_loop_breaker(3)
try:
    breaker.guard("search", {"q": "refund"}, lambda: run_search("refund"))
except RetryLoopError:
    # the agent was looping — break out and re-plan
    ...

Async jobs

cave.jobs targets /api/v1/jobs on control_url and falls back to base_url. submit() returns a Job, status() re-reads it, wait() polls to a terminal state, submit_and_wait() does both, and cancel() requests cancellation. This is for gateway deployments that expose that public API path.

job = cave.jobs.submit_and_wait(
    {"kind": "report.export", "project_id": project_id},
    latency_class="background",
    timeout_s=120.0,
)
print(job.state)  # "completed" | "failed" | "cancelled"

See also