canary-ops-sdk 0.1.0

The Operating System for AI Security — monitor, secure, and audit every AI agent and LLM application in production.

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.

---

Install

pip install canary-ops-sdk

No runtime dependencies. Python 3.8+.

---

Quick start

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

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

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

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

# 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

# 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

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

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