cohesion-sdk 1.2.0

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 Judgment Independence Score API.

COHESION is the global certification standard for human oversight of AI. The API scores whether humans are genuinely exercising judgment over AI outputs, and produces the compliance proof regulators require (EU AI Act Article 14, Colorado SB 205, NYC Local Law 144, SR 11-7).

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

Install

pip install cohesion-sdk

Requires Python 3.11 or newer.

Quickstart

Get a key:

• New customer: sign up at https://cohesionauth.com/signup (self-serve; pay + receive key in under 5 minutes)
• 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.1.0)

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.

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.