keel-sdk

Python SDK for the Keel managed execution and AI Permit APIs.

On its managed paths, Keel controls whether consequential AI actions may execute before they happen and records evidence afterward. This SDK's provider wrappers use Keel-managed proxy execution.

Keel is live with self-serve access. Create an account and project API key in the dashboard. Production deployment assistance is available from Keel. Get started →

Install

pip install keel-sdk

Quick Start — One-Line Migration

Add Keel governance to your existing AI code with a single import change. No other code modifications needed.

OpenAI:

# BEFORE:
from openai import OpenAI

# AFTER:
from keel_sdk.providers.openai import OpenAI

client = OpenAI()  # Uses KEEL_BASE_URL, KEEL_API_KEY, KEEL_PROJECT_ID env vars
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
    max_tokens=100,
)
print(response["choices"][0]["message"]["content"])

Anthropic:

# BEFORE:
from anthropic import Anthropic

# AFTER:
from keel_sdk.providers.anthropic import Anthropic

client = Anthropic()  # Uses KEEL_BASE_URL, KEEL_API_KEY, KEEL_PROJECT_ID env vars
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    messages=[{"role": "user", "content": "Hello!"}],
    max_tokens=1024,
)
print(response["content"][0]["text"])

Google (Gemini):

# BEFORE:
from google.generativeai import GenerativeModel

# AFTER:
from keel_sdk.providers.google import GenerativeModel

model = GenerativeModel("gemini-2.0-flash")
response = model.generate_content("Hello!")
print(response["candidates"][0]["content"]["parts"][0]["text"])

xAI (Grok):

# BEFORE:
from openai import OpenAI; client = OpenAI(base_url="https://api.x.ai/v1", api_key="xai-...")

# AFTER:
from keel_sdk.providers.xai import Grok

client = Grok()
response = client.chat.completions.create(
    model="grok-2",
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response["choices"][0]["message"]["content"])

Meta (Llama):

# BEFORE:
from openai import OpenAI; client = OpenAI(base_url="https://api.llama-api.com", ...)

# AFTER:
from keel_sdk.providers.meta import Llama

client = Llama()
response = client.chat.completions.create(
    model="llama-3.3-70b",
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response["choices"][0]["message"]["content"])

Every call goes through Keel's proxy for that provider: Keel issues a Permit (policy and budget check), dispatches the call only if the Permit allows it, records actual usage from the provider response, and keeps audit evidence.

Request Lifecycle

Every request processed by Keel follows a consistent high-level flow:

• Evaluate: identity, policy, and budget constraints are checked
• Decide: a permit decision is issued: allow, deny, review (held for human approval), or throttle
• Execute: the provider call occurs only if permitted
• Record: usage, cost, and governance events are captured

Requests are only executed if explicitly permitted.

Provider Wrappers

The provider wrappers in keel_sdk.providers give you drop-in replacements for the official OpenAI, Anthropic, Google (Gemini), xAI (Grok), and Meta (Llama) SDKs. They do not depend on or import the provider packages. Instead, they talk directly to Keel's proxy endpoints.

Configuration

Set environment variables or pass explicitly:

from keel_sdk.providers.openai import OpenAI

# Via env vars (recommended):
#   KEEL_BASE_URL=https://api.keelapi.com
#   KEEL_API_KEY=keel_sk_...
#   KEEL_PROJECT_ID=<your project id, a UUID>
client = OpenAI()

# Or pass explicitly:
client = OpenAI(
    keel_base_url="https://api.keelapi.com",
    keel_api_key="keel_sk_...",
    keel_project_id="0b0b0b0b-4444-4c4c-8d8d-999900001111",
    keel_subject={"type": "user", "id": "usr_42"},  # optional, defaults to service/default
)

The api_key parameter (for OpenAI compatibility) is accepted but ignored — Keel manages provider keys. keel_project_id must be your project's UUID; a different project id is answered with 404 project_not_found.

Governance Flow

On every create() call, the wrapper sends one request, POST /v1/proxy/openai (or /anthropic, /google, /xai, /meta), with the provider-shaped body plus your project id, subject and any parent Permit or session id, under a fresh Idempotency-Key. Keel's proxy then:

1. Permits the call (policy and budget check).
2. Refuses it when the Permit does not allow it: the wrapper raises KeelError (ThrottledError for HTTP 429) with the refusal's code, for example policy.rule_denied, and the Permit id in e.permit_id.
3. Dispatches the call to the provider.
4. Records actual usage and evidence from the provider response.

What each proxy forwards:

• Anthropic forwards model, max_tokens, text messages, system, temperature and top_p. The wrapper sends system= as a leading system message and joins text-only content blocks into a string. Tool use (tools, tool_choice, tool_use/tool_result blocks) and image blocks raise UnsupportedByProxyError before anything is sent: use managed execution for those. stop_sequences and top_k are not forwarded, so they are not sent (you get a one-time warning).
• Google requires contents as a list of Content objects: generate_content("Hello!") and lists of strings are sent as one user Content. generation_config is sent as generationConfig with camelCase keys.

Keel has marked the OpenAI and Anthropic proxy routes deprecated in favor of managed execution (/v1/execute). When a response carries Deprecation: true, the SDK emits one DeprecationWarning per route per process naming the sunset date and the successor.

Async Support

from keel_sdk.providers.openai import AsyncOpenAI
from keel_sdk.providers.anthropic import AsyncAnthropic
from keel_sdk.providers.google import AsyncGenerativeModel
from keel_sdk.providers.xai import AsyncGrok
from keel_sdk.providers.meta import AsyncLlama

client = AsyncOpenAI()
response = await client.chat.completions.create(model="gpt-4o", messages=[...])

client = AsyncAnthropic()
response = await client.messages.create(model="claude-sonnet-4-20250514", messages=[...], max_tokens=1024)

model = AsyncGenerativeModel("gemini-2.0-flash")
response = await model.generate_content_async("Hello!")

client = AsyncGrok()
response = await client.chat.completions.create(model="grok-2", messages=[...])

client = AsyncLlama()
response = await client.chat.completions.create(model="llama-3.3-70b", messages=[...])

Streaming

Pass stream=True to get an iterator of chunks:

for chunk in client.chat.completions.create(model="gpt-4o", messages=[...], stream=True):
    print(chunk)

An execution refusal returned as a complete execution envelope yields terminal execution.denied or execution.error events followed by done. Other non-2xx streaming responses raise a typed KeelError. When retries are enabled, executions.stream() retries a transient refusal before any stream events arrive; it never replays a stream after returning events.

Error Handling

from keel_sdk import KeelError, ThrottledError
from keel_sdk.providers.openai import OpenAI

client = OpenAI()
try:
    response = client.chat.completions.create(model="gpt-4o", messages=[...])
except ThrottledError as e:
    print(f"Throttled ({e.code}); retry in {e.retry_after_seconds}s")
except KeelError as e:
    if e.status == 403:
        print(f"Blocked by Keel: {e.code} ({e.message}); permit {e.permit_id}")
    else:
        print(f"API error {e.status}: {e.message}")

Setup

from keel_sdk import KeelClient

client = KeelClient(
    base_url="https://api.keelapi.com",
    api_key="keel_sk_...",
)

Advanced external enforcement (permit-first)

Most integrations should use the managed provider wrappers or execution clients above. Use the Permit API directly only when another trusted system owns enforcement and execution:

import uuid
from keel_sdk import KeelClient

client = KeelClient(base_url="https://api.keelapi.com", api_key="keel_sk_...")
PROJECT_ID = "0b0b0b0b-4444-4c4c-8d8d-999900001111"  # your project's UUID

permit = client.permits.create({
    "project_id": PROJECT_ID,
    "idempotency_key": str(uuid.uuid4()),
    "subject": {"type": "user", "id": "usr_123"},
    "action": {"name": "ai.generate"},
    "resource": {
        "type": "request",
        "id": "req_123",
        "attributes": {
            "provider": "openai",
            "model": "gpt-4o-mini",
            "estimated_input_tokens": 200,
            "estimated_output_tokens": 500,
        },
    },
    # Caller-defined context is kept with the Permit, and policy conditions
    # can read it (context.*).
    "context": {"ticket": "SUP-123"},
})

if permit["decision"] == "allow":
    # The trusted external enforcement point may proceed.
    pass

permit["decision"] is one of allow, deny, review (held for an exact human approval) or throttle; permit["reason_code"] says why. A platform precondition (plan quota, billing state, an unknown workflow) refuses issuance with HTTP 403, raised as PermitStructuralRefusalError (e.outcome_kind == "precondition").

Async variant:

permit = await client.permits.create_async({...})

create_with_meta() returns the Permit together with the governance metadata Keel stamped in the response headers:

result = client.permits.create_with_meta({...})
result["permit"]["permit_id"], result["request_id"], result["governance"].decision

Agent request proofs

An agent that holds a registered credential can sign the exact Permit request. Pass the proof headers with the call; the SDK sends them unchanged and does not replace a timestamp or nonce you supply:

permit = client.permits.create(request, headers={
    "X-Keel-Agent-Credential-Id": credential_id,
    "X-Keel-Agent-Signature": signature,
    "X-Keel-Timestamp": timestamp,
    "X-Keel-Nonce": nonce,
})

Dry run

result = client.permits.dry_run(permit_request)

List and get

permits = client.permits.list(project_id=PROJECT_ID, decision="review", limit=50)
permit = client.permits.get("permit_id")
envelope = client.permits.get("permit_id", view="envelope")

Report usage

Usage close-out needs an admin-scope API key, verification material (caller_report is refused) and a positive cost:

client.permits.report_usage("permit_id", {
    "actual_input_tokens": 180,
    "actual_output_tokens": 420,
    "cost_usd_micros": 1_850,
    "verification": {
        "method": "provider_receipt",
        "provider_request_id": "chatcmpl-abc123",
        "receipt_json": {"id": "chatcmpl-abc123", "model": "gpt-4o-mini", "usage": {...}},
    },
})
client.permits.verify_usage("permit_id", {"method": "provider_receipt", "status": "verified", "provider_request_id": "chatcmpl-abc123"})

Attestation, evidence, lineage, lifecycle

# Approve (or reject) a Permit held for review; needs an approval-scope key.
client.permits.attest("permit_id", {"attestation_type": "approve", "rationale": "Checked the invoice."})
client.permits.add_evidence("permit_id", {
    "evidence_type": "hash",
    "evidence_value": "sha256:abc123",
    "label": "response_hash",
    "attached_by": "billing-service",
})
evidence = client.permits.list_evidence("permit_id")
lineage = client.permits.lineage("permit_id")
timeline = client.permits.timeline("permit_id")        # this Permit's lifecycle
bundle = client.permits.bundle("permit_id", profile="permit-exact.v3")
receipt = client.permits.receipt("permit_id")
proof_pack = client.permits.proof_pack("permit_id")    # ZIP archive bytes

Workflows

Declare a multi-call workflow before it runs, then use the workflow context manager to thread X-Keel-Workflow-Id through every SDK call made inside the block.

from keel_sdk import WORKFLOW_INTENT_MAX_CALLS_EXCEEDED, KeelClient

client = KeelClient(base_url="https://api.keelapi.com", api_key="keel_sk_...")

decl = client.workflows.declare(
    "invoice-batch-2026-05-13",
    {
        "expected_calls": 10000,
        "max_calls": 12000,
        "expected_model": "gpt-5-mini",
        "expected_input_tokens_per_call": 4000,
        "expected_output_tokens_per_call": 500,
        "max_duration_seconds": 86400,
    },
)

with client.workflow(decl["workflow_id"]):
    permit = client.permits.create({
        "project_id": PROJECT_ID,
        "idempotency_key": "invoice-001",
        "subject": {"type": "service_account", "id": "billing-agent"},
        "action": {"name": "ai.generate"},
        "resource": {
            "type": "request",
            "id": "invoice-001",
            "attributes": {
                "provider": "openai",
                "model": "gpt-5-mini",
                "estimated_input_tokens": 4000,
                "estimated_output_tokens": 500,
            },
        },
    })

if permit["decision"] == "deny" and permit["reason_code"] == WORKFLOW_INTENT_MAX_CALLS_EXCEEDED:
    # The workflow reached max_calls; amend it or stop the run.
    client.workflows.amend(
        decl["workflow_id"],
        if_match_version=decl["version"],
        new_max_calls=20000,
        reason_provided="ticket volume higher than forecast",
    )
client.workflows.complete(decl["workflow_id"])

Async calls use the same context variable:

async with client.workflow_async("invoice-batch-2026-05-13"):
    permit = await client.permits.create_async({...})

Reaching max_calls is a Permit decision (deny with reason workflow_intent.max_calls_exceeded). Workflow API errors are typed subclasses of KeelError, including WorkflowAmendmentVersionConflictError, WorkflowUnknownOrInactiveError and PlanUpgradeRequiredError. The proxy routes (and the provider wrappers) raise WorkflowMaxCallsExceededError for the same limit; an MCP call raises McpToolDeniedError whose reason_code names it.

A fuller runnable sample lives in examples/workflows/declare_and_run.py.

Executions

Run a model synchronously. Managed execution issues its own Permit:

result = client.executions.create({
    "operation": "generate.text",
    "messages": [{"role": "user", "content": "Summarize this document."}],
    "routing": {"provider": "openai", "model": "gpt-4o-mini"},
})

A refused or failed execution is returned as an execution envelope with status and error rather than raised.

Stream tokens as they arrive:

for event in client.executions.stream({
    "operation": "generate.text",
    "messages": [{"role": "user", "content": "Write a poem."}],
}):
    if event["event_type"] == "content.delta":
        print(event["data"]["delta"]["text"], end="", flush=True)
    elif event["event_type"] == "execution.completed":
        print()
        usage = event["data"]["usage"]
    elif event["event_type"] in ("execution.denied", "execution.error"):
        print(event["data"]["error"]["message"])

Async streaming:

async for event in client.executions.stream_async({...}):
    ...

executions.create, executions.stream and execute.run accept an idempotency_key. With automatic retries enabled, the SDK generates one per call so a retried request is never executed twice.

Execute (unified)

result = client.execute.run({
    "model": "gpt-4o-mini",
    "provider": "openai",
    "input": {"messages": [{"role": "user", "content": "Translate to Spanish: Hello world"}]},
})

For value movement, set "action_verb": "payment.execute" and pass an idempotency_key.

Proxy

Pass requests through to providers with Keel governance applied:

response = client.proxy.openai(
    {"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "Hello"}]},
    idempotency_key="order-42",
)

# Also: client.proxy.anthropic(), .google(), .xai(), .meta()

MCP

Govern Model Context Protocol tool calls. Server management, decide and get_decision_record need an admin-scope key; prepare_tool_call and call_tool need a client key bound to an agent.

from keel_sdk import (
    McpActionReviewRequiredError,
    McpPreconditionRefusedError,
    McpStructuralHoldError,
    McpToolDeniedError,
    ThrottledError,
)

try:
    result = client.mcp.call_tool(
        "payments", "transfer", {"arguments": {"amount": 100}}, idempotency_key="transfer-42"
    )
except McpActionReviewRequiredError as e:
    # An exact human approval may release this request: approve e.permit_id,
    # then retry with the same idempotency key.
    print(e.permit_id, e.permit_ref, e.permit_status)
except McpToolDeniedError as e:
    print("denied:", e.reason_code)
except McpStructuralHoldError as e:
    print("Keel could not establish a precondition:", e.reason_code)
except McpPreconditionRefusedError as e:
    print("platform precondition:", e.code, e.outcome_kind)
except ThrottledError as e:
    print("throttled; retry in", e.retry_after_seconds, "s")

A refused key answers IdempotencyKeyRefusedError (409): send a new key. Top-level metadata keys starting with _keel_ and the names agent_principal_shadow and shadow_override are reserved; Keel refuses them with 400 mcp.reserved_metadata_key.

Framework integrations

keel_sdk.integrations.langchain.KeelChatModel (pip install 'keel-sdk[langchain]') and keel_sdk.integrations.llamaindex.KeelLLM (pip install 'keel-sdk[llamaindex]') acquire a Keel Permit before every call of the model they wrap, thread parent/child Permits across nested calls, and carry the active Langfuse trace id in the Permit's context. A throttle decision raises ThrottledError; any other refusal raises KeelError. They do not close out usage: that needs an admin-scope key, verification material and the call's cost, so report it with client.permits.report_usage() if you settle usage yourself.

Jobs

Submit async jobs and poll for results. A job carries the Permit request it runs under:

job = client.jobs.create({
    "permit": {
        "project_id": PROJECT_ID,
        "idempotency_key": "job-001",
        "subject": {"type": "service_account", "id": "batch-worker"},
        "action": {"name": "ai.generate"},
        "resource": {
            "type": "request",
            "id": "job-001",
            "attributes": {
                "provider": "openai",
                "model": "gpt-4o-mini",
                "estimated_input_tokens": 2000,
                "estimated_output_tokens": 500,
            },
        },
    },
    "provider_payload": {"messages": [{"role": "user", "content": "Analyze this dataset."}]},
})

status = client.jobs.get(job["job_id"])
# status["status"]: "submitted" | "queued" | "processing" | "completed" | "failed"

API Keys

key = client.api_keys.create()
keys = client.api_keys.list()
client.api_keys.revoke(key["id"])

create() answers HTTP 202 with a PendingApiKeyCreateResponse (status: "requested") when the new key needs approval.

Request Timeline

timeline = client.requests.timeline("request_id")

Error Handling

from keel_sdk import KeelClient, KeelError, PermitStructuralRefusalError, ThrottledError

try:
    client.permits.create(request)
except ThrottledError as e:
    # HTTP 429 (subclass of KeelError)
    print(e.code)                 # e.g. "budget.rate_limit_throttled" or "rate_limited"
    print(e.retry_after_seconds)  # Retry-After, else error.details.retry_after_seconds
    print(e.reason_code)
    print(e.permit_id)            # when the throttle names a Permit
except PermitStructuralRefusalError as e:
    # HTTP 403: a platform precondition refused issuance; no Permit exists
    print(e.reason_code, e.outcome_kind)
except KeelError as e:
    print(e.status)        # HTTP status code
    print(e.code)          # e.g. "invalid_request", "forbidden", "policy.rule_denied"
    print(e.message)       # human-readable message
    print(e.field)         # field that caused the error, if any
    print(e.details)       # error details
    print(e.body)          # the full parsed response body
    print(e.headers)       # the response's x-keel-* headers
    print(e.permit_id)     # the Permit the response names, if any
    print(e.request_id)    # x-keel-request-id, when present
    print(e.is_retryable)  # True for 408, 429, 500, 502, 503, 504
    print(e.retry_after)   # seconds from Retry-After header, or None

Keel's errors use an {"error": {"code", "message", ...}} envelope. A few routes answer {"detail": ...}: a token such as project_not_found becomes the code, and a validation list becomes validation_error. A body that is not JSON gets the code http_<status>.

Throttle (HTTP 429)

When Keel throttles a request it returns HTTP 429 with an error envelope such as {"error": {"code": "budget.rate_limit_throttled", "message": ..., "details": {"retry_after_seconds": 30}}} and usually a Retry-After header. The SDK:

1. Reads Retry-After (falls back to retry_after_seconds in the body).
2. Auto-retries if retry_config is set (429 is in the default retryable codes), under a fresh Idempotency-Key: Keel treats a throttled request's key as spent.
3. Raises ThrottledError after retries are exhausted (or immediately if retries are disabled).

A Permit decided as throttle (HTTP 200) is not an error on permits.create(); the framework integrations raise ThrottledError for it.

Reason Codes

Refusals carry a dot-namespaced reason_code. The SDK exports every code in Keel's reason-code lexicon (ALL_REASON_CODES) plus POLICY_EXPLICIT_AUTHORIZATION_REQUIRED:

from keel_sdk import BUDGET_RATE_LIMIT_THROTTLED, POLICY_MODEL_NOT_ALLOWED

if e.reason_code == BUDGET_RATE_LIMIT_THROTTLED:
    ...

Refusal bodies also say whether a rule decided the outcome: outcome_kind is decision when a rule ran and precondition when Keel could not evaluate the request as configured.

Automatic Retries

The SDK can automatically retry failed requests that return transient HTTP errors (408, 429, 500, 502, 503, 504) using exponential backoff with jitter.

from keel_sdk import KeelClient, RetryConfig

# Use default retry settings (3 retries, 0.5s initial delay, 2x backoff)
client = KeelClient(
    base_url="https://api.keelapi.com",
    api_key="keel_sk_...",
    retry_config=RetryConfig(),
)

# Customize retry behavior
client = KeelClient(
    base_url="https://api.keelapi.com",
    api_key="keel_sk_...",
    retry_config=RetryConfig(
        max_retries=5,
        initial_delay=1.0,
        max_delay=60.0,
        backoff_multiplier=3.0,
    ),
)

To disable retries (the default), omit retry_config or pass None:

client = KeelClient(
    base_url="https://api.keelapi.com",
    api_key="keel_sk_...",
    retry_config=None,  # no automatic retries
)

When a response includes a Retry-After header, the SDK respects that value instead of the computed backoff delay. A POST that carries an Idempotency-Key is retried under the same key after 408 or a 5xx, and under a fresh key after 429. Streaming calls are never retried.

Per-Request Timeout

KeelClient.request() and request_async() accept an optional timeout that overrides the client-level timeout for that single request; the sub-client methods use the client-level timeout.

# Client-level timeout is 30 s (the default)
client = KeelClient(base_url="https://api.keelapi.com", api_key="keel_sk_...", timeout=30.0)

# This specific call uses a 5 s timeout instead
permit = client.request("POST", "/v1/permits", json=body, timeout=5.0)

Pass timeout=None (or omit it) to use the client-level default.

Context Manager

with KeelClient(base_url="...", api_key="...") as client:
    permit = client.permits.create({...})

# Async
async with KeelClient(base_url="...", api_key="...") as client:
    permit = await client.permits.create_async({...})

Freshness Headers

X-Keel-Timestamp and X-Keel-Nonce are sent on every request by default, including from the provider wrappers. Hosted Keel and any deployment running with a non-dev APP_ENV require them on execution routes and on POST /v1/permits; without them those routes return 401 request_not_fresh. Deployments that do not enforce freshness ignore the headers.

A fresh nonce is generated per attempt, so automatic retries are not rejected as replays. A timestamp or nonce you pass in headers= is sent unchanged.

# Default: headers are sent.
client = KeelClient(base_url="https://api.keelapi.com", api_key="keel_sk_...")

# Opt out only when an intermediary rejects unknown request headers.
client = KeelClient(
    base_url="https://api.keelapi.com",
    api_key="keel_sk_...",
    request_freshness=False,
)

Governance Metadata

Every response carries the decision metadata Keel stamps in x-keel-* response headers, so you can get the permit id for a specific call without a second request. The response is a dict subclass, so it behaves exactly as before.

response = client.chat.completions.create(...)

response["choices"][0]["message"]["content"]  # unchanged
response.keel.permit_id                       # the Permit for this call
response.keel.decision                        # allow | deny | review | throttle
response.keel.reason
response.keel.request_id                      # quote this in support requests

response.keel.present is False when Keel stamped no governance headers — for example on health and management routes.

Health

client.get_health()         # GET /health, no key needed
client.get_authed_health()  # GET /v1/health, validates the API key

License

MIT