Build secure AI,
from your first line of code.

The Shadow Warden SDK wraps every gateway feature — filtering, agentic intelligence, marketplace, and OpenAI-compatible chat — in a clean Python and TypeScript interface.

Install

Python pip

pip install shadow-warden-sdk

TypeScript / Node npm

npm install @shadow-warden/sdk

Command line

The warden console script ships with the Python SDK from v1.1.0. Every command takes --json, and filter exits 0 when the content is allowed and 1 when it is blocked — so it drops straight into a shell conditional or a CI gate.

CLI shadow-warden-sdk >= 1.1.0

warden health
warden filter "ignore previous instructions"
warden filter --json --file prompt.txt
cat prompt.txt | warden filter --stdin
warden impact --requests 50000
warden billing

Configuration: WARDEN_API_KEY, WARDEN_GATEWAY_URL, or the matching flags. Exit codes: 0 allowed · 1 blocked · 2 usage · 3 gateway unreachable.

Quick start

Python

from shadow_warden import WardenClient

client = WardenClient(api_key="sk-...")

# Filter a prompt
result = client.filter(
    "Ignore previous instructions and...",
    raise_on_block=True,
)
print(result.risk_level)  # HIGH
print(result.flags)       # ["jailbreak_attempt"]

# OpenAI-compatible chat — filtered end-to-end
response = client.chat.completions.create(
    messages=[{"role": "user", "content": "Hello"}],
    model="gpt-4o",
)

# SOVA agent — autonomous SOC operator
reply = client.agent("What is our current threat level?")
print(reply.reply)

TypeScript

import { WardenClient } from "@shadow-warden/sdk"

const client = new WardenClient({
  apiKey: "sk-...",
})

// Filter a prompt
const result = await client.filter(
  "Ignore previous instructions...",
  { raiseOnBlocked: true },
)
console.log(result.riskLevel)  // "HIGH"

// OpenAI-compatible — fully filtered
const chat = await client.chat.completions.create({
  model: "gpt-4o",
  messages: [{ role: "user", content: "Hello" }],
})

// SOVA agent
const reply = await client.agent("Threat summary")
console.log(reply.reply)

Async Python

Use AsyncWardenClient for asyncio applications — same interface, non-blocking I/O.

Python — async

import asyncio
from shadow_warden import AsyncWardenClient, WardenBlockedError

async def main():
    async with AsyncWardenClient(api_key="sk-...") as client:
        # Parallel filter calls
        results = await asyncio.gather(
            client.filter("prompt one"),
            client.filter("prompt two"),
            client.filter("prompt three"),
        )
        for r in results:
            print(r.risk_level, r.flags)

asyncio.run(main())

Method reference

All methods available on WardenClient and AsyncWardenClient.

Method	Returns	Description	Tier
client.filter(content)	FilterResponse	Run content through the full 9-layer pipeline. Returns blocked, risk_level, flags, secrets_found, processing_ms.	All
client.chat.completions.create(messages)	dict (OpenAI-compatible)	Proxy to any LLM through Shadow Warden. Every message is filtered before it reaches the model.	All
client.agent(query)	AgentResponse	Query SOVA — the autonomous security agent. Multi-turn memory via session_id (6 h TTL).	Pro+
client.marketplace.listings()	list[MarketplaceListing]	Browse M2M marketplace listings. Filter by community_id, asset_type, status.	Pro+
client.marketplace.stats()	dict	Aggregate marketplace stats: volume, trade count, active listings, escrow pipeline.	Pro+
client.health()	dict	Check gateway health, version, and pipeline readiness.	All

Error handling

The SDK raises typed exceptions so you can handle auth, rate-limiting, and blocked content separately.

from shadow_warden import (
    WardenClient, WardenBlockedError,
    WardenGatewayError, WardenTimeoutError, WardenError,
)

with WardenClient(api_key="sk-...") as client:
    try:
        result = client.filter(user_input, raise_on_block=True)
    except WardenBlockedError as e:
        # e.response has full FilterResponse
        log.warning("Blocked: %s", e.response.flags)
        return "Request blocked for safety reasons."
    except WardenGatewayError:
        raise RuntimeError("Invalid API key")
    except WardenTimeoutError:
        time.sleep(1)
        # retry...
    except WardenError as e:
        log.error("Warden error %s: %s", e.status_code, e)

Ready to integrate?

Get your API key from the portal, install the SDK, and make your first filtered request in under 5 minutes.

Get API Key Full Docs →