cohesion-sdk 1.3.2

Official Python SDK for the COHESION Judgment Independence Score API.

cohesion-sdk (Python)

Save humanity by keeping judgment alive in the age of AI.

Official Python SDK for the COHESION API. COHESION measures whether human oversight of AI is real, durable, and reviewable. The SDK wraps the live scoring, routing, audit-chain, and compliance-rollup endpoints; pair it with an OpenAI, Anthropic, Azure OpenAI, Bedrock, Vertex, or in-house model to instrument every human-AI interaction.

Reference scopes a buyer typically maps the evidence model to: EU AI Act Article 14, Colorado SB 205, Colorado SB 26-189 ADMT, NYC Local Law 144, SR 11-7.

Full product documentation: https://cohesionauth.com/docs API reference: https://cohesionauth.com/api

Evaluate COHESION in 30 minutes

The canonical buyer-engineer evaluation path. Each step has a runnable artifact.

1. Curl the unauthenticated root: curl -s https://api.cohesionauth.com/v1 | jq.
2. Get a key at https://cohesionauth.com/pricing (Starter $499/mo via the pricing page; FDP fit-call gated).
3. pip install cohesion-sdk.
4. Score one decision (see Quickstart below).
5. Replay it via curl against the live endpoint: curl https://api.cohesionauth.com/v1/decision/replay/$AI_DECISION_LOG_ID -H "X-API-Key: $COHESION_API_KEY" (no SDK wrapper).
6. Pull a compliance rollup: client.compliance_report().
7. Verify the audit chain via curl (admin-only, master-key auth, no SDK wrapper): curl https://api.cohesionauth.com/v1/admin/audit/verify-chain -H "X-API-Key: $COHESION_ADMIN_API_KEY".
8. Open https://dashboard.cohesionauth.com and search by request_id.

If any step stalls, email [e-mail] with the failing step and the request_id from the SDK error.

JIS vs DRS, when to use each

Engine	Question it answers	Subject of measurement	Primary endpoint
JIS (Judgment Independence Score)	Is the reviewer still exercising independent judgment?	The reviewer, scored against their own history	POST /v1/score
DRS (Decision Risk Score)	Is this particular decision risky enough to require a human, an async review, or a block?	The individual decision, scored against policy	POST /v1/decision/score

DRS routes the decision; JIS scores whether the reviewer is a trustworthy router.

Pilot workflow: score one AI decision end-to-end

The same loop every production integration runs.

1. Operator opens a task; your app records session_id + operator_id.
2. Your app calls the AI provider.
3. Your UI presents the AI recommendation; the SDK instruments the interaction in the background.
4. The operator commits a decision.
5. Your app calls client.score(...); SDK attaches Idempotency-Key automatically.
6. The engine returns the JIS envelope under 100ms p50.
7. Your app stores request_id from the response against the decision. If you are using the DRS gate (client.decision_score() / gate()), also store the ai_decision_log_id it returns. That is the identifier GET /v1/decision/replay/{ai_decision_log_id} accepts.

Generate an oversight evidence packet

The artifact a design partner gets at pilot close. Two SDK calls plus one curl assemble it. The audit-chain verifier is admin-only (master-key auth) and is not wrapped by the SDK; call it directly with curl.

report  = client.compliance_report()
profile = client.operator_profile("analyst-42")

# Admin-only -- uses the scoped master key, NOT the customer scoring key.
curl https://api.cohesionauth.com/v1/admin/audit/verify-chain \
  -H "X-API-Key: $COHESION_ADMIN_API_KEY"

Contents: org-level oversight band distribution, reviewer-level JIS band over time, dimension breakdowns, decay projection, and a tamper-evident audit chain.

Human oversight evidence model

What the SDK produces, conceptually:

1. Observable telemetry from the human-AI interaction (latency, hover, scroll, alternative views, decision class, modification extent).
2. A decision-level score (DRS) that routes the decision BEFORE it reaches the end-user.
3. A reviewer-level score (JIS) that scores whether the reviewer is still functioning as an oversight signal over time.
4. A tamper-evident audit chain that connects telemetry, scores, and routing into one verifiable record per decision.
5. An evidence packet assembled from the rollup endpoints.

Invisible-by-design: no extension, no survey, no offline assessment.

Example: healthcare intake workflow

A clinical intake nurse uses an AI scribe to draft the chief complaint.

client.score(
    operator_id="rn-22871",
    session_id="intake_2026_05_27_0094",
    domain="healthcare",
    interaction={
        "ai_recommendation_presented": True,
        "time_to_decision_ms": 41200,
        "decision": "modified",
        "modification_extent": 0.45,
        "ai_available": True,
        "scenario_type": "standard",
        "outcome_correct": None,
        "hover_events": 6,
        "scroll_depth": 0.92,
        "alternative_views_checked": 2,
    },
)

Example: lending decision workflow

A loan officer reviews an AI-recommended credit decision. DRS gates first, then JIS scores the reviewer interaction.

from cohesion import gate

pre = await gate(client, request, None)
if pre["routing_decision"] != "auto":
    return route_to_reviewer(pre)

ai_output = await provider.complete(request)
await gate(client, request, ai_output)  # post-AI strict gate

client.score(
    operator_id="officer-104",
    session_id="loan_8821",
    domain="financial",
    interaction={...},
)

Example: HR screening workflow

A recruiter screens AI-shortlisted resumes under NYC Local Law 144. Every accept/reject is logged as an oversight event.

client.score(
    operator_id="recruiter-58",
    session_id="screen_2026_05_27_0044",
    domain="general",
    interaction={
        "ai_recommendation_presented": True,
        "time_to_decision_ms": 27800,
        "decision": "rejected",
        "modification_extent": 0,
        "ai_available": True,
        "scenario_type": "standard",
        "outcome_correct": None,
        "hover_events": 4,
        "scroll_depth": 0.7,
        "alternative_views_checked": 3,
    },
)

When the city audit window opens, client.compliance_report() returns the aggregate; the admin-only GET /v1/admin/audit/verify-chain endpoint (curl, master-key auth) proves the chain is unbroken.

Install

pip install cohesion-sdk

Requires Python 3.11 or newer.

Quickstart

Get a key:

• New customer: start at https://cohesionauth.com/pricing. Dev is free (raw JSON responses only). Starter is $499/mo or $4,990/yr. The Founding Design Partner cohort is $4,900 one-time, credited 100 percent toward the first year of any annual plan, founder-reviewed via a brief fit call; the cohort closes 2026-07-15. Audited is $1,999/mo or $19,900/yr. Enterprise starts at $50K and is scoped procurement, set up with the founder. Once checkout is live, a completed checkout provisions an org and delivers your API key by email. Key shown once, then rotate from your dashboard.
• Existing customer rotating a key: https://cohesionauth.com/dashboard/api-keys

Then:

from cohesion import Client

client = Client(api_key="ck_live_...your_key...")

response = client.score(
    session_id="sess-001",
    operator_id="analyst-42",
    domain="financial",
    interaction={
        "ai_recommendation_presented": True,
        "time_to_decision_ms": 1800,
        "decision": "modified",
        "modification_extent": 0.3,
        "ai_available": True,
        "scenario_type": "standard",
        "outcome_correct": None,
        "hover_events": 2,
        "scroll_depth": 0.7,
        "alternative_views_checked": 1,
    },
)

print(response.jis, response.band)

DRS gate -- the Authorized Inference Gateway

cohesion.gate() wraps a regulated AI call with the Decision Risk Score engine. The contract is strict-gate, fail-closed: the SDK refuses to surface a final AI output unless DRS returned a clean routing_decision envelope.

from cohesion import (
    Client,
    gate,
    CohesionPolicyBlockedError,
    CohesionRequiresHumanReviewError,
    CohesionMalformedDecisionError,
)

client = Client(api_key=os.environ["COHESION_API_KEY"])

# 1. Pre-AI pass: ask DRS whether the input is admissible at all.
pre = await gate(client, request, None)
if pre["routing_decision"] != "auto":
    # must_review / async_review / forced_escalation -- route to a human
    # reviewer BEFORE calling the AI provider. pre["review_queue_id"] is
    # the routing handle.
    return route_to_reviewer(pre)

# 2. Call the AI provider.
ai_output = await provider.complete(request)

# 3. Post-AI pass: ask DRS whether to surface the AI output.
try:
    post = await gate(client, request, ai_output)
    # routing_decision == 'auto', no forced escalation -- safe to surface.
    return ai_output
except CohesionPolicyBlockedError as e:
    # Hard block. e.decision["drs_reason_codes"] explains why.
    return block_response(e)
except CohesionRequiresHumanReviewError as e:
    # Route to reviewer. e.decision["review_queue_id"] is the handle.
    return route_to_reviewer(e.decision)
except CohesionMalformedDecisionError as e:
    # Fail-closed sentinel: 2xx but envelope malformed. Treat as hard block.
    return block_response(e)
# Any other CohesionError / ServerError / NetworkError (network / 5xx /
# retry-exhausted) propagates unchanged -- caller MUST treat as fail-closed.

Error class taxonomy:

Exception	Raised on	Carries
CohesionPolicyBlockedError	routing_decision == 'policy_blocked' on either pass	decision["drs_reason_codes"], decision["policy_evaluation"]["hard_failures"]
CohesionRequiresHumanReviewError	Post-AI must_review / async_review / non-empty forced_escalation_rules_triggered	decision["review_queue_id"]
CohesionMalformedDecisionError	2xx with missing or unrecognized envelope fields (fail-CLOSED)	observed_keys, missing_fields
ServerError / NetworkError	retry-exhausted transport failure	request_id

Pre-AI must_review / async_review / forced escalation return the envelope without raising -- that branch is the caller's signal to route to a reviewer before the AI call.

See examples/07_live_gate.py for a runnable end-to-end demonstration.

Same call, three languages

cURL

curl -X POST https://api.cohesionauth.com/v1/score \
  -H "X-API-Key: ck_live_...your_key..." \   # gitleaks:allow (placeholder in doc example)
  -H "Content-Type: application/json" \
  -d '{
    "session_id": "sess-001",
    "operator_id": "analyst-42",
    "domain": "financial",
    "interaction": {
      "ai_recommendation_presented": true,
      "time_to_decision_ms": 1800,
      "decision": "modified",
      "modification_extent": 0.3,
      "ai_available": true,
      "scenario_type": "standard",
      "outcome_correct": null,
      "hover_events": 2,
      "scroll_depth": 0.7,
      "alternative_views_checked": 1
    }
  }'

Python

from cohesion import Client
client = Client(api_key="ck_live_...your_key...")
resp = client.score(session_id="sess-001", operator_id="analyst-42", domain="financial", interaction={...})

TypeScript

import { Cohesion } from "@cohesionauth/sdk";
const client = new Cohesion({ apiKey: "ck_live_...your_key..." });
const resp = await client.score({ session_id: "sess-001", operator_id: "analyst-42", domain: "financial", interaction: {...} });

5-minute integrations (1.3.1)

Three drop-in patterns. Pick the one that fits your stack.

FastAPI / Starlette ASGI middleware

import os
from fastapi import FastAPI
from cohesion import Client
from cohesion.fastapi import CohesionMiddleware

cohesion = Client(api_key=os.environ["COHESION_API_KEY"])

app = FastAPI()
app.add_middleware(
    CohesionMiddleware,
    client=cohesion,
    domain="general",  # SOC alert-triage maps here per the v1.2 spec
    operator_id_resolver=lambda req: req.headers.get("x-user-id", "anon"),
)

Decorator (any function -- sync or async)

from cohesion import Client, measure

cohesion = Client(api_key=os.environ["COHESION_API_KEY"])

@measure(
    client=cohesion,
    domain="financial",
    operator_id=lambda *_a, **kw: kw["user_id"],
)
async def handle_request(prompt: str, *, user_id: str) -> str:
    ...

domain is one of healthcare | aviation | financial | legal | pharmaceutical | general. SOC use cases map to general per v1.2 §4 Option A.

For chat-completion patterns (OpenAI / Anthropic / LangChain), see Provider wrappers below or import the chat namespace:

from cohesion import chat
wrapped = chat.wrap_openai(openai_client, cohesion, ctx)

Six full integration examples live under examples/.

Client surface

client = Client(
    api_key: str,                             # required
    base_url: str = "https://api.cohesionauth.com",
    timeout: float = 30.0,
    max_retries: int = 3,
    logger: logging.Logger | None = None,
    enable_telemetry: bool = False,
)

client.score(**kwargs)                    # POST /v1/score
client.score_batch(interactions)          # POST /v1/score/batch (<= 100 per call)
client.operator_profile(operator_id)      # GET /v1/operator/:id/profile
client.organization_dashboard()           # GET /v1/organization/dashboard
client.maintenance_recommend(**kwargs)    # POST /v1/maintenance/recommend
client.compliance_report()                # GET /v1/compliance/report
client.admin_key_rotate()                 # POST /v1/admin/key/rotate
client.admin_key_revoke()                 # POST /v1/admin/key/revoke
client.admin_audit_log(event_type=None, since=None, until=None, limit=100)

Provider wrappers

Wrap any OpenAI, Anthropic, or Azure OpenAI client to emit oversight telemetry for free.

from openai import OpenAI
from cohesion import Client, WrapContext, DecisionReport, wrap_openai

cohesion = Client(api_key="ck_live_...")
openai_client = OpenAI()
wrapped = wrap_openai(
    openai_client,
    cohesion,
    WrapContext(operator_id="analyst-42", session_id="sess-001", domain="financial"),
)

completion = wrapped.chat.completions.create(model="gpt-4", messages=[...])
# Your app logic: analyst reviews the AI suggestion and commits a decision
completion.record_decision(DecisionReport(decision="modified", modification_extent=0.3))

LangChain users:

from cohesion import LangChainCallback
cb = LangChainCallback(
    client=cohesion,
    operator_id="analyst-42",
    session_id="sess-001",
    domain="financial",
)
chain.invoke(inputs, config={"callbacks": [cb]})

CLI

export COHESION_API_KEY=ck_live_...your_key...
cohesion score --operator-id analyst-42 --from-file payload.json

# Read payload from stdin:
cat payload.json | cohesion score --operator-id analyst-42 --from-file -

Errors

Every exception carries:

• request_id (from the API response, None for local validation)
• next_step (actionable remediation; None when the SDK cannot know what to do)

Exception	When it fires	next_step populated
AuthenticationError	HTTP 401	Yes. Names the dashboard URL and rotation path.
RateLimitError	HTTP 429	Yes. "Retry after N seconds." retry_after field also set.
ValidationError	HTTP 400/413/422 or local checks	Yes when field is known: "field must be one of: ..."
ServerError	HTTP 5xx	No (not actionable client-side)
NetworkError	Transport / timeout	Yes when resolvable (connectivity, TLS egress).
CohesionError	Base class	Depends

from cohesion import CohesionError, RateLimitError

try:
    client.score(...)
except RateLimitError as err:
    print(err.next_step, err.retry_after, err.request_id)
except CohesionError as err:
    print(err.next_step, err.request_id)

Retry and idempotency

• Exponential backoff with full jitter, max 3 attempts by default (max_retries).
• Retries on 429 and 5xx only. Honors RFC-7231 Retry-After (integer seconds or HTTP-date).
• Every POST auto-generates an Idempotency-Key header (UUIDv4) so safe retries do not create duplicate records.

SDK parity matrix

Every documented endpoint is reachable from the Python SDK, the TypeScript SDK, and the dashboard. Operator-only paths (key rotation, audit log inspection) are read-write in the SDKs and read in the dashboard.

Endpoint	Python	TypeScript	Dashboard
POST /v1/score	client.score()	client.score()	read
POST /v1/score/batch	client.score_batch()	client.scoreBatch()	read
POST /v1/decision/score	client.decision_score() / gate()	client.decisionScore() / gate()	read
GET /v1/decision/:id	client.get_decision()	client.getDecision()	read
GET /v1/decision/queue	client.decision_queue()	client.decisionQueue()	read
GET /v1/decision/replay/:id	curl (no SDK wrapper)	curl (no SDK wrapper)	read
GET /v1/operator/:id/profile	client.operator_profile()	client.operatorProfile()	read
GET /v1/organization/dashboard	client.organization_dashboard()	client.organizationDashboard()	read
POST /v1/maintenance/recommend	client.maintenance_recommend()	client.maintenanceRecommend()	read
GET /v1/compliance/report	client.compliance_report()	client.complianceReport()	read + export
POST /v1/telemetry/ai-call-observed	client.telemetry_ai_call_observed()	client.telemetryAiCallObserved()	n/a
GET /v1/admin/audit/verify-chain	curl (master-key, no SDK wrapper)	curl (master-key, no SDK wrapper)	read
POST /v1/admin/key/rotate	client.admin_key_rotate()	client.adminKeyRotate()	read + rotate
POST /v1/admin/key/revoke	client.admin_key_revoke()	client.adminKeyRevoke()	read
GET /v1/admin/audit-log	client.admin_audit_log()	client.adminAuditLog()	read

Full 50-endpoint catalog at the OpenAPI schema.

Endpoint status classification

Every endpoint is tagged with one of four maturity classifications.

Class	Meaning	Backward compatibility
Production	Stable contract. SLA-backed.	12-month deprecation notice via Sunset response header.
Beta	Public, intentionally exposed, contract may change.	Notice via release notes; no Sunset guarantee.
Admin	Org-scoped, owner-only. Key rotation, revocation, audit log.	Production-grade backward compatibility.
Internal	Reserved for the dashboard + SDK.	No external contract.

The Production class covers the entire surface listed in the parity matrix above. Admin endpoints are tagged in the OpenAPI schema with x-cohesion-class: admin.

Errors and recovery

Every failure mode has a documented recovery path. Catch by class, never by string match.

Failure	HTTP	SDK class	Caller next step
Missing or wrong key	401	AuthenticationError	Check the key prefix. Rotate at dashboard.cohesionauth.com/api-keys.
Bad payload field	400 / 413 / 422	ValidationError	Inspect err.field and err.allowed_values.
Rate limit hit	429	RateLimitError	SDK auto-retries with backoff + jitter; honor err.retry_after if surfaced after retries.
Upstream model unavailable (5xx)	503 / 504	ServerError	SDK retries on 5xx; after retries, fail-closed (do NOT surface the AI output).
Network down / DNS / TLS	transport	NetworkError	Verify egress to api.cohesionauth.com. Fail-closed.
DRS envelope malformed	2xx	CohesionMalformedDecisionError	Treat as hard block. Contact support with err.request_id.

Every exception carries request_id and next_step. The CohesionError base is the safe fallback catch.

Compliance boundaries

What the SDK helps customers prove, and what it does NOT prove.

The SDK measures:

• Whether a named human reviewer was present at the moment of decision.
• Whether the reviewer exercised independent judgment, or rubber-stamped.
• Whether the decision was risky enough to require an additional human, an async review, or a hard block.
• Whether the reviewer's oversight quality is degrading over time.
• Whether the entire interaction is reconstructable from a tamper-evident audit chain.

The SDK does NOT:

• Render legal opinions or determine regulatory conformance on a customer's behalf.
• Replace internal model-risk-management or model-validation processes.
• Score the underlying AI model's correctness or bias.
• Operate as a certification body.

Buyers and their counsel are responsible for mapping the evidence model to specific regulatory obligations.

Data handling and retention

• Transport: TLS 1.3 from your environment to api.cohesionauth.com.
• Storage region: Cloudflare D1 in the customer's contracted region (EU customers on EU edge; US customers on US edge).
• Retention windows: Telemetry and score envelopes: 13 months default, configurable per contract. Audit chain: indefinite (a chain cannot be truncated without breaking proof).
• Redaction: The default logger redacts API keys (pattern ck_live_*) and operator_id values before emit. Free-text prompt content is NEVER stored by the engine.
• Export: client.compliance_report() + client.admin_audit_log() are SDK-callable and downloadable.

Audit trail example

A synthetic but realistic excerpt of the chain returned by GET /v1/admin/audit/verify-chain (admin-only, master-key auth, curl-only -- not wrapped by the SDK). Three consecutive decisions; the chain head proves none were silently dropped or rewritten.

{
  "chain_head": "sha256:a13c…f902",
  "verified": true,
  "entries": [
    {
      "request_id": "req_01HXG7K9Z2W4M6P8B0A2C4E6F8",
      "timestamp": "2026-05-27T14:08:42.318Z",
      "operator_id_hash": "h_e2a9…",
      "decision_class": "modified",
      "drs_routing": "auto",
      "jis_band": "Strong",
      "prev_hash": "sha256:9b71…a004",
      "this_hash": "sha256:c84d…ee19"
    },
    {
      "request_id": "req_01HXG7KCV4Y2T8M0R6E3K1Q9P2",
      "timestamp": "2026-05-27T14:10:01.044Z",
      "operator_id_hash": "h_e2a9…",
      "decision_class": "rejected",
      "drs_routing": "must_review",
      "jis_band": "Strong",
      "prev_hash": "sha256:c84d…ee19",
      "this_hash": "sha256:1f02…7733"
    },
    {
      "request_id": "req_01HXG7KFM8N0Q3D5W2A8H6Y1B6",
      "timestamp": "2026-05-27T14:11:55.610Z",
      "operator_id_hash": "h_e2a9…",
      "decision_class": "accepted",
      "drs_routing": "auto",
      "jis_band": "Adequate",
      "prev_hash": "sha256:1f02…7733",
      "this_hash": "sha256:a13c…f902"
    }
  ]
}

Each entry's this_hash derives from its content plus the previous entry's hash. The chain head is the cryptographic summary of every decision in-window.

Logging and redaction

The SDK logs under the cohesion logger at INFO by default. JSON format when COHESION_LOG_FORMAT=json. A deny-list filter redacts:

• Any string matching ck_live_[A-Za-z0-9]+ (replaced with ck_live_[redacted])
• Any record field whose key is one of: api_key, x-api-key, authorization, operator_id, new_api_key

API keys and operator_id values never reach the log sink.

Opt-in telemetry

client = Client(api_key="...", enable_telemetry=True)
# Requires:  pip install cohesion-sdk[telemetry]
# And set COHESION_SENTRY_DSN or SENTRY_DSN.

Only SDK-level errors are reported. The PII scrubber strips request body, headers, and operator_id fields.

Environment variables

Var	Purpose
COHESION_API_KEY	Picked up by the CLI (not by Client() directly)
COHESION_BASE_URL	Override base URL
COHESION_TEST_API_KEY	Gates the pytest -m integration live test
COHESION_LOG_FORMAT	Set to json for JSON logs
COHESION_SENTRY_DSN	Overrides SENTRY_DSN when telemetry is enabled

Versioning and support

Semver. Breaking changes ship with a 12-month notice via Deprecation + Sunset response headers. See CHANGELOG.md.

Security issues: [e-mail].

License

Apache-2.0. See LICENSE and NOTICE.