cohesion-sdk 1.1.0

Official Python SDK for the COHESION Judgment Independence Score API.

cohesion-sdk (Python)

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 at 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)

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: {...} });

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.