Python SDK

pip install klent-sdk

Requires Python ≥ 3.10. Single dependency: httpx.

Most apps want a higher-level entry point. This page documents the raw HTTP-shaped methods (start_execution, evaluate_action, log_event). For day-to-day code, the auto-instrumentation helpers — run_tool, run_anthropic_agent — collapse the five-step boilerplate into a single call and run a complete Anthropic tool-use loop end-to-end. See Auto-instrumentation.

Instantiating the client

from klent_sdk import KlentClient

klent = KlentClient(api_key=os.environ["KLENT_API_KEY"])

Arguments

Context manager

Prefer the context-manager form in short-lived scripts so buffered events flush cleanly:

with KlentClient(api_key=os.environ["KLENT_API_KEY"]) as klent:
    execution = klent.start_execution({"agent_id": "billing-agent"})
    ...
# Events are flushed on __exit__.

Methods

start_execution(body)

execution = klent.start_execution({
    "agent_id": "support-agent",
    "metadata": {"user_id": "u_123"},
})
# → Execution TypedDict

evaluate_action(body)

decision = klent.evaluate_action({
    "execution_id": execution["id"],
    "tool": "send_email",
    "input": {"to": "[email protected]", "subject": "Hi"},
})

if decision["decision"] == "deny":
    raise RuntimeError(decision["reason"])

log_event(body)

Non-blocking — buffered on a background queue.

klent.log_event({
    "execution_id": execution["id"],
    "type": "action_executed",
    "payload": {"tool": "send_email"},
})

get_pending_action(pending_action_id, *, wait_ms=0)

Fetch the current state of a pending action — the row created when an approve policy parks an action for human review. Returns a typed dict with status (pending, approved, rejected) plus, on terminal states, reviewer note and any last-mile modifications.

row = klent.get_pending_action("pact_…")

if row["status"] == "pending":
    # still waiting on a human
    ...

wait_ms opts into a server-side long-poll: the request stays open until the row resolves or the budget elapses (server caps each call at 30 000 ms; run_tool's approval_wait chains long-polls under the hood for longer budgets).

row = klent.get_pending_action("pact_…", wait_ms=30_000)

Most apps don't call this directly — run_tool(... approval_wait={"timeout_seconds": 600}) in Auto-instrumentation wraps it. Use the raw method when you're polling from a different process (e.g. a worker that resumes paused executions).

flush()

Force-flush the buffer. Called automatically via atexit and on context-manager exit.

klent.flush()

close()

Flush and close the underlying HTTP client. Also called by __exit__.

klent.close()

Thread safety

EventBuffer is thread-safe under concurrent log_event calls and serializes flushes via an inner lock. You can share a single KlentClient across threads without wrapping.

Error handling

Methods raise:

• ValueError — missing api_key.
• RuntimeError — a 4xx from the API (not retried) or exhausted retries on 5xx/429.
• httpx.HTTPError subclasses only if you set max_retries=0 and the underlying network call fails.

The message includes the HTTP status and the response body so you can diagnose without reaching for a debugger.

Types

TypedDicts live in klent_sdk.types:

from klent_sdk.types import (
    CreateExecutionRequest,
    EvaluateActionRequest,
    EvaluateActionResponse,
    Event,
    EventType,
    Execution,
    LogEventRequest,
    PolicyEffect,
)

They match the TypeScript SDK types field-for-field.