Metadata-Version: 2.4
Name: canary-ops-sdk
Version: 0.1.0
Summary: The Operating System for AI Security — monitor, secure, and audit every AI agent and LLM application in production.
Project-URL: Homepage, https://canary-ops.base44.app
Project-URL: Repository, https://github.com/CryptoSuess/canary-sdk-python
Project-URL: Bug Tracker, https://github.com/CryptoSuess/canary-sdk-python/issues
Project-URL: Changelog, https://github.com/CryptoSuess/canary-sdk-python/blob/main/CHANGELOG.md
Author: CryptoSuess & Txnchi
License: MIT
Keywords: agent-monitoring,ai-firewall,ai-security,canary,llm,llmops,observability,prompt-injection,telemetry
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.8
Description-Content-Type: text/markdown

# canary-ops-sdk

**The Operating System for AI Security** — Python SDK for monitoring, securing, and auditing every AI agent and LLM application in production.

Zero-dependency telemetry, redaction, and firewall SDK. Works with any Python LLM library.

[![PyPI](https://img.shields.io/pypi/v/canary-ops-sdk)](https://pypi.org/project/canary-ops-sdk/)
[![Python](https://img.shields.io/pypi/pyversions/canary-ops-sdk)](https://pypi.org/project/canary-ops-sdk/)

---

## Install

```bash
pip install canary-ops-sdk
```

No runtime dependencies. Python 3.8+.

---

## Quick start

```python
from canary_ops import Canary

canary = Canary(
    api_key="sk-canary-...",   # from your Canary dashboard
    agent_id="my-agent",
)

# Manual tracking
canary.track_prompt("What is the capital of France?", model="gpt-4o")
canary.track_response("Paris.", model="gpt-4o", tokens_in=10, tokens_out=3)
canary.track_tool_call("web_search", {"query": "Paris"}, {"results": [...]})

canary.flush()  # or call shutdown() at process exit
```

---

## OpenAI instrumentation

```python
from openai import OpenAI
from canary_ops import Canary

canary = Canary(api_key="sk-canary-...", agent_id="my-agent")
client = canary.wrap_openai(OpenAI())

# All chat completions are now tracked automatically
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
)
```

Streaming is also supported — just pass `stream=True`.

---

## Anthropic instrumentation

```python
import anthropic
from canary_ops import Canary

canary = Canary(api_key="sk-canary-...", agent_id="my-agent")
client = canary.wrap_anthropic(anthropic.Anthropic())

response = client.messages.create(
    model="claude-opus-4-5",
    messages=[{"role": "user", "content": "Hello!"}],
    max_tokens=256,
)
```

---

## Prompt firewall

```python
from canary_ops import Canary, CanaryBlockedError, ScanOptions

canary = Canary(api_key="sk-canary-...", agent_id="my-agent")

# scan() — returns a ScanResult, never raises
result = canary.scan(user_prompt, ScanOptions(remote=False))
if not result.allowed:
    return "I can't help with that."

# guard() — raises CanaryBlockedError if the prompt is blocked
try:
    safe_prompt = canary.guard(user_prompt)
except CanaryBlockedError as e:
    print(f"Blocked: {e.scan_result.threats}")
```

---

## Output scanning

```python
# Scan a response for leaked secrets, PII, or exfiltration
from canary_ops import ScanResponseOptions

result = canary.scan_response(
    llm_output,
    ScanResponseOptions(throw_on_unsafe=True),
)
```

---

## Canary tokens

```python
# Plant a decoy credential in your agent's context
token = canary.plant_token()
system_prompt = f"Company API key: {token.token_value}\n{your_context}"

# If the token appears in any response, Canary fires a data_exfiltration alert
result = canary.scan_response(response_text)
```

---

## RAG context scanning

```python
from canary_ops import ScanContextOptions

docs = retriever.get_relevant_documents(query)
results = canary.scan_context(
    [d.page_content for d in docs],
    ScanContextOptions(concurrency=5),
)
safe_docs = [d for d, r in zip(docs, results) if r.allowed]
```

---

## Configuration reference

```python
from canary_ops import Canary, GuardConfig, RedactionConfig, RetryConfig

canary = Canary(
    api_key="sk-canary-...",
    agent_id="my-agent",
    # Privacy
    capture_content=True,          # set False for metadata-only mode
    max_content_length=10_000,     # truncate long prompts
    redaction=RedactionConfig(
        secrets=True,              # strip API keys, tokens, private keys
        pii=True,                  # strip emails, phones, SSNs, credit cards
        mode="mask",               # "mask" | "hash" | "remove"
    ),
    sample_rate=0.5,               # only send 50% of events
    # Firewall
    guard=GuardConfig(
        local=True,                # run in-process heuristics
        remote=True,               # call the remote deep scanner
        block_on_local=True,       # short-circuit on high-confidence local hit
        responses=False,           # auto-scan every LLM response
    ),
    fail_open=True,                # allow prompts if the remote scanner is unreachable
    # Resilience
    timeout_ms=10_000,
    retry=RetryConfig(retries=3, backoff_ms=250),
    # Batching
    flush_interval_ms=5_000,       # drain queue every 5 s
    max_batch_size=20,             # flush immediately when queue hits 20
    # Hooks
    before_send=lambda event: event,   # inspect/mutate/drop events
    on_error=lambda exc: print(exc),   # custom error handler
    # Tagging
    environment="production",
    release="v1.2.3",
    debug=False,
)
```

---

## Error reference

| Exception | When it's raised |
|---|---|
| `CanaryBlockedError` | `guard()` or `scan_response(..., throw_on_unsafe=True)` when the firewall blocks |
| `CanaryApiError` | `plant_token()` when the API call fails |
| `CanaryNotInitializedError` | Calling SDK methods before `Canary(...)` is constructed |

`CanaryBlockedError` carries a `.scan_result` attribute with the full `ScanResult`.

---

## Async support

All wrapper patching detects whether the underlying client's `create` method is a coroutine and patches it with an async wrapper automatically. The `Canary` client itself is thread-safe; background flushing runs on a daemon thread.

---

## License

MIT © CryptoSuess & Txnchi
