Python SDK cookbook

Concrete recipes for middleware, batch, retries, and async concurrency.

Middleware around an AI call

import time
from cohesion import Client

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

def ai_with_cohesion(prompt, operator_id, session_id):
    started = time.time()
    ai_out = call_your_ai_provider(prompt)
    elapsed_ms = int((time.time() - started) * 1000)

    decision = capture_operator_decision(ai_out)

    cohesion.score(
        session_id=session_id,
        operator_id=operator_id,
        domain="financial",
        interaction={
            "ai_recommendation_presented": True,
            "time_to_decision_ms": elapsed_ms,
            "decision": decision.kind,
            "modification_extent": decision.edit_ratio,
            "ai_available": True,
            "scenario_type": "standard",
            "outcome_correct": None,
            "hover_events": decision.hovers,
            "scroll_depth": decision.scroll,
            "alternative_views_checked": decision.alt_views,
        },
    )
    return ai_out

Batch backfill from a telemetry store

from itertools import islice

def chunk(iterable, n):
    it = iter(iterable)
    while batch := list(islice(it, n)):
        yield batch

for interactions in chunk(telemetry.stream(), 100):
    cohesion.score_batch(operator_id="analyst-42", domain="financial", interactions=interactions)

Retry on 429

The SDK retries automatically with exponential backoff. To customize:

client = Client(api_key="...", max_retries=5)

Async concurrency

import asyncio
from cohesion import AsyncClient

async def score_many(items):
    async with AsyncClient(api_key=os.environ["COHESION_API_KEY"]) as c:
        return await asyncio.gather(*(c.score(**item) for item in items))

Idempotency keys

The SDK auto-generates a UUIDv4 Idempotency-Key on every POST. No manual step needed. Server-side deduplication is live. If you need to correlate a score call with a prior telemetry beacon, pass request_id as a keyword argument:

import uuid
cohesion.score(..., request_id=str(uuid.uuid4()))

Next step