TheWARDN Python SDK

Official Python SDK for TheWARDN governance-as-a-service platform. Govern your AI agents with Sentinel rules, CHAM policies, and escrow-based human-in-the-loop review.

Installation

pip install thewardn

Quick Start

from thewardn import TheWARDN

wardn = TheWARDN(api_key="wdn_live_...")
result = wardn.govern(
    agent_id="my-agent",
    action_type="deploy",
    target_service="api-server",
)
if result.execute:
    deploy()  # Action cleared by governance

Authentication

TheWARDN uses two authentication mechanisms:

• API Key (X-WARDN-KEY header) -- used by the /govern endpoint. This is what your agents use at runtime.
• JWT Token (Authorization: Bearer header) -- used by management endpoints (agents, escrow). This is what your dashboard/admin tools use.

# Agent runtime -- API key only
wardn = TheWARDN(api_key="wdn_live_...")

# Admin operations -- API key + JWT
wardn = TheWARDN(api_key="wdn_live_...", jwt_token="eyJ...")

API Reference

TheWARDN(api_key, base_url, timeout, max_retries, jwt_token)

govern(agent_id, action_type, target_service, ...)

Submit an action for governance review. Returns a GovernResult.

result = wardn.govern(
    agent_id="brainiac-001",
    action_type="restart_service",
    target_service="payment-api",
    reasoning="Service health check failed 3 times",
    confidence={"incident": 0.95, "fix": 0.88, "containment": 0.92},
    environment="production",
    metadata={"ticket": "INC-4521"},
)

print(result.verdict)    # "CLEARED", "HELD", or "BLOCKED"
print(result.execute)    # True if CLEARED
print(result.tier)       # "A", "B", "C", or "X"
print(result.escrow_id)  # Set if HELD (use to release/kill later)

Parameters:

GovernResult

Agent Management (requires JWT)

# List agents
agents = wardn.list_agents()

# Register a new agent
agent = wardn.register_agent(
    name="BrA.Iniac",
    agent_type="incident-response",
    who_i_am="I am a governed AI agent for production incident response.",
)

# Get agent details
agent = wardn.get_agent("agent-id")

# Update agent
wardn.update_agent("agent-id", name="New Name")

# Deregister agent
wardn.delete_agent("agent-id")

Escrow Management (requires JWT)

# List held items
items = wardn.get_escrow()

# Get specific item
item = wardn.get_escrow_item("escrow-id")

# Approve a held action
wardn.release_escrow("escrow-id")

# Deny a held action
wardn.kill_escrow("escrow-id")

# Queue stats
stats = wardn.escrow_stats()

Async Support

The SDK includes a fully async client for use with asyncio:

import asyncio
from thewardn import AsyncTheWARDN

async def main():
    async with AsyncTheWARDN(api_key="wdn_live_...") as wardn:
        result = await wardn.govern(
            agent_id="my-agent",
            action_type="deploy",
            target_service="api-server",
        )
        if result.execute:
            await deploy()

asyncio.run(main())

All methods on AsyncTheWARDN mirror TheWARDN but return coroutines.

Error Handling

from thewardn import TheWARDN, AuthenticationError, RateLimitError, GovernanceError, ConnectionError

wardn = TheWARDN(api_key="wdn_live_...")

try:
    result = wardn.govern(
        agent_id="my-agent",
        action_type="deploy",
        target_service="api-server",
    )
except AuthenticationError:
    print("Invalid API key")
except RateLimitError:
    print("Monthly limit exceeded -- upgrade your plan")
except GovernanceError:
    print("Governance engine unavailable -- fail safe, do not proceed")
except ConnectionError:
    print("Cannot reach TheWARDN API")

All exceptions inherit from WARDNError and include:

• .message -- human-readable error
• .status_code -- HTTP status code (if applicable)
• .response -- raw error response (if applicable)

Retry Logic

The SDK automatically retries transient failures (HTTP 502, 503, 504, connection errors, timeouts) with exponential backoff. Default is 3 retries with 0.5s base delay (0.5s, 1s, 2s).

# Custom retry config
wardn = TheWARDN(
    api_key="wdn_live_...",
    max_retries=5,
    timeout=60,
)

Using with the Confidence Dataclass

from thewardn import TheWARDN, Confidence

wardn = TheWARDN(api_key="wdn_live_...")
result = wardn.govern(
    agent_id="my-agent",
    action_type="deploy",
    target_service="api-server",
    confidence=Confidence(incident=0.95, fix=0.88, containment=0.92),
)

Thread Safety

The synchronous TheWARDN client is thread-safe and uses connection pooling via httpx.Client. You can share a single instance across threads.

License

MIT -- see LICENSE.