Metadata-Version: 2.4
Name: cos-sdk
Version: 0.6.0
Summary: 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.
Author-email: "Protofine.ai" <support@protofine.ai>
License-Expression: MIT
Project-URL: Homepage, https://protofine.ai
Project-URL: Documentation, https://cos.protofine.ai/console
Project-URL: Repository, https://github.com/protofine/cos-python-sdk
Keywords: ai,accountability,governance,audit,hallucination,cos,openai,gemini,claude,wrapper
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.24.0
Provides-Extra: openai
Requires-Dist: openai>=1.0; extra == "openai"
Provides-Extra: gemini
Requires-Dist: google-generativeai>=0.3.0; extra == "gemini"
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.20.0; extra == "anthropic"
Provides-Extra: all
Requires-Dist: openai>=1.0; extra == "all"
Requires-Dist: google-generativeai>=0.3.0; extra == "all"
Requires-Dist: anthropic>=0.20.0; extra == "all"
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Dynamic: license-file

# 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](https://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

```bash
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

```python
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:

```python
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:

```python
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

```python
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:

```python
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](https://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):

```python
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

```python
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)

```python
# 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

```python
# 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

```python
# 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

```python
# 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.

```bash
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`.

```bash
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](https://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

```python
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")
```
