cos-sdk 0.6.0

Python SDK for COS (Cognitive Operating System) by Protofine.ai — the accountability and control layer between any AI model and production. Drop-in wrappers for OpenAI, Gemini, and Claude.

COS Python SDK

Python client for the COS (Cognitive Operating System) API — the accountability and control layer between any AI model and production, by Protofine.ai.

Your models generate; COS judges what becomes action. Every AI response gets a confidence score, a risk level, and flagged claims with corrections — and every verdict is backed by a signed audit receipt you can show a third party. Deletion requests can be dispatched, and verified, across every LLM that absorbed your data.

Drop-in wrappers for OpenAI, Gemini, and Claude — add COS to your existing AI pipeline with one line of code.

Install

pip install cos-sdk

# With LLM wrapper support (pick what you use):
pip install cos-sdk[openai]      # OpenAI (GPT-4, GPT-3.5)
pip install cos-sdk[gemini]      # Google Gemini
pip install cos-sdk[anthropic]   # Anthropic Claude
pip install cos-sdk[all]         # All wrappers

Quick Start

from cos_sdk import COS

client = COS(api_key="cos_live_xxx")

# Validate AI-generated text
result = client.validate(
    text="According to a 2024 study, 90% of Fortune 500 companies use AI in production.",
    tier="tier_2",  # tier_1 (instant), tier_2 (AI review), tier_3 (multi-model)
)

print(f"Confidence: {result.confidence_score}")  # 0.0 to 1.0
print(f"Risk: {result.risk_level}")              # "low", "medium", "high"

for claim in result.flagged_claims:
    print(f"  Flag: {claim.claim}")
    print(f"  Why:  {claim.reason}")
    if claim.correction:
        print(f"  Fix:  {claim.correction}")

Streaming

Get results progressively — Tier 1 arrives in ~5ms, deeper tiers in 2-5 seconds:

for event in client.validate_stream("AI text to check", tier="tier_3"):
    if event.event == "t1_result":
        print(f"Quick check: {event.result.confidence_score}")
    elif event.event == "validation_complete":
        print(f"Full result: {event.result.confidence_score}")

Inline sentence-level warnings (Cyna sentinel, v0.6+)

With tier="cyna", the stream emits one sentence_validated event per sentence in the AI response — perfect for inline chat UIs that color-highlight flagged sentences as each verdict arrives:

for event in client.validate_stream(ai_response, tier="cyna"):
    if event.event == "sentence_validated":
        sv = event.sentence  # SentenceVerdict dataclass
        if sv.flagged:
            print(f"⚠️ [{sv.index}] {sv.text}")
            print(f"   {sv.flag.reason}")
    elif event.event == "validation_complete":
        # Aggregated verdict across all sentences
        print(f"Overall: {event.result.summary}")

Sentences run through Cyna in parallel server-side, so N sentences take ~2s wall clock (not N × 2s). Flag severity is low/medium/high.

Async

from cos_sdk import AsyncCOS

async with AsyncCOS(api_key="cos_live_xxx") as client:
    result = await client.validate("AI text to check")
    print(result.confidence_score)

Zero-SDK: point the OpenAI client at COS (universal proxy)

If you already use the OpenAI SDK, you don't need cos-sdk at all. Swap base_url to COS and every chat completion gets routed to the right provider, validated, and audit-receipted:

from openai import OpenAI

client = OpenAI(
    api_key="cos_live_xxx",           # your COS API key
    base_url="https://cos.protofine.ai",
)

# Models route by prefix:
#   gpt-* / o1-* / o3-* / o4-* → OpenAI (your OpenAI key required)
#   claude-*                   → Anthropic (your Anthropic key required)
#   gemini-*                   → Gemini (COS-billed, no customer key)
#   mistral-* / ministral-*    → Mistral (your Mistral key required)

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Who invented Python?"}],
)

print(resp.choices[0].message.content)

# COS validation metadata lives in response headers:
#   x-cos-receipt-id    — signed audit receipt ID
#   x-cos-confidence    — 0.000–1.000
#   x-cos-risk-level    — low | medium | high
#   x-cos-cache         — hit | miss | skip | error
#   x-cos-parse-warning — present when JSON-mode / tool_calls didn't parse

Register provider keys

Customer provider keys live on your COS API key doc — not in request headers. Add them at cos.protofine.ai/console → Provider Keys. A missing key returns 402 with a pointer to the console.

Per-request knobs (optional)

Pass a cos object in the request body (COS-only; providers ignore it):

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[...],
    extra_body={
        "cos": {
            "tier": "tier_3",                    # tier_1/tier_2/tier_3/bamboo/cyna/auto
            "mode": "guard",                     # or "monitor" (async, zero-latency)
            "include_validation_in_body": True,  # mirrors metadata into body.cos_validation
        },
    },
)

Monitor mode — zero-latency validation

resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[...],
    extra_body={"cos": {"mode": "monitor"}},
)
# Response returns immediately. Fetch the verdict later via
# /api/v2/monitor/{id} using the x-cos-monitor-id header.

Streaming

stream=True is supported for gpt-* and mistral-* today; claude-* and gemini-* return 501 with a clear message (streaming normalization lands in a later commit). A COS-namespaced SSE event with the monitor ID arrives right before the terminal [DONE].

---

Drop-in LLM Wrappers (v0.4.0)

Add COS to your existing AI calls — change one import, everything else stays the same.

OpenAI (GPT-4, GPT-3.5)

# Before — no validation
from openai import OpenAI
client = OpenAI()

# After — every response validated by COS
from cos_sdk.openai import COS_OpenAI
client = COS_OpenAI(cos_api_key="cos_live_xxx")

response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "What is quantum computing?"}]
)

# Normal OpenAI response — works exactly the same
print(response.choices[0].message.content)

# COS trust score — attached automatically
print(response.cos_validation.confidence_score)  # 0.95
print(response.cos_validation.risk_level)         # "low"

Google Gemini

# Before
import google.generativeai as genai
model = genai.GenerativeModel("gemini-1.5-flash")

# After
from cos_sdk.gemini import COS_Gemini
model = COS_Gemini("gemini-1.5-flash", cos_api_key="cos_live_xxx")

response = model.generate_content("Explain black holes")
print(response.text)                                # Normal Gemini response
print(response.cos_validation.confidence_score)      # COS trust score

Anthropic Claude

# Before
from anthropic import Anthropic
client = Anthropic()

# After
from cos_sdk.anthropic import COS_Anthropic
client = COS_Anthropic(cos_api_key="cos_live_xxx")

message = client.messages.create(
    model="claude-3-haiku-20240307",
    max_tokens=1024,
    messages=[{"role": "user", "content": "What is quantum computing?"}]
)
print(message.content[0].text)                       # Normal Claude response
print(message.cos_validation.confidence_score)        # COS trust score

Block High-Risk Responses

# Automatically raise an error if the AI hallucinates
client = COS_OpenAI(
    cos_api_key="cos_live_xxx",
    cos_block_high_risk=True,  # Raises HighRiskResponseError
    cos_tier="bamboo",         # Use Bamboo (best accuracy)
)

Resilience

If COS is down, your AI still works. COS validation is non-blocking — if the COS API is unreachable, cos_validation is set to None and the original AI response passes through untouched.

Memory hygiene — request a forget across every LLM (L8.3)

When an end-user (or a regulator under GDPR/HIPAA/SOC 2) demands their data be erased, COS dispatches a deletion request to every LLM that may have absorbed it and returns one signed attestation per LLM. Opt-in per API key (forget_enabled=True); defaults off.

curl -X POST https://cos.protofine.ai/api/v2/memory/forget \
  -H "Authorization: Bearer cos_live_xxx" \
  -H "Content-Type: application/json" \
  -d '{
    "content_description": "Delete all references to project Falcon",
    "target_llms": ["gemini-workspace", "chatgpt-memory", "claude-projects", "custom-endpoint"],
    "org_id": "your-org",
    "customer_forget_url": "https://your-self-hosted-shim.example/forget"
  }'

Returns { job_id, attestations[] } — every attestation has a receipt_id you can show a regulator. One LLM's outage cannot crash the call: the other targets still attest. Re-fetch the job with GET /api/v2/memory/forget/{job_id}.

Verify the LLM actually forgot — canary probes (L8.4)

Adding canary_probes_enabled=True to your API key tells COS to follow every forget call with indirect probe queries that try to surface the deleted content. The probes run as a background task; the POST returns immediately. Poll the job state to see verification results — canary_probe_results[] is a typed list, each probe carrying its attempt (1, 2, or 3 — escalating retry), residual_detected flag, and signed receipt_id. Job-level status settles on one of pending, verified, residual_detected, or skipped.

curl -H "Authorization: Bearer cos_live_xxx" \
  https://cos.protofine.ai/api/v2/memory/forget/$JOB_ID
# → { "job_id", "attestations":[…], "canary_probe_results":[{ "attempt": 1, "residual_detected": false, … }], "status": "verified", "completed_at": "…" }

If status="residual_detected" after the BG task completes, COS already retried the forget 3× with escalating descriptions ("…including paraphrases", "…all conversation context referring to…") before giving up. Surface the failure to the regulator with the full probe receipt chain attached.

Validation Tiers

Tier	Speed	Cost	What it does
tier_1	~5ms	Free	Heuristic pattern matching (fake stats, suspicious URLs, hedging)
tier_2	~2s	Low	Single-model AI review
tier_3	~3s	Medium	Multi-model consensus
bamboo	~3s	Low	Purpose-built fine-tuned model (deepest analysis)

Get Your API Key

1. Go to cos.protofine.ai/console
2. Sign in with your account
3. Create a new API key
4. Save it — you won't see it again

Error Handling

from cos_sdk import COS, AuthenticationError, RateLimitError

client = COS(api_key="cos_live_xxx")

try:
    result = client.validate("text to check")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Slow down — retry in {e.retry_after}s")