thewardn 0.1.0

Python SDK for TheWARDN governance-as-a-service platform

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)

Parameter	Type	Default	Description
api_key	str	required	Your WARDN API key
base_url	str	https://api.thewardn.ai	API base URL
timeout	int	30	Request timeout (seconds)
max_retries	int	3	Retries for transient 502/503/504 errors
jwt_token	str	None	JWT for management endpoints

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:

Parameter	Type	Default	Description
agent_id	str	required	Registered agent ID
action_type	str	required	Action being taken (e.g. "deploy", "restart")
target_service	str	required	Target service name
reasoning	str	None	Why the agent wants to do this
confidence	Confidence or dict	None	Confidence scores (incident/fix/containment, 0-1)
environment	str	"production"	Target environment
metadata	dict	None	Arbitrary metadata

GovernResult

Field	Type	Description
.verdict	str	"CLEARED", "HELD", or "BLOCKED"
.execute	bool	True if CLEARED
.cleared	bool	Alias for .execute
.held	bool	True if HELD in escrow
.blocked	bool	True if BLOCKED
.tier	str	Governance tier (A/B/C/X)
.escrow_id	str	Escrow ID if held
.seq	int	Audit sequence number
.hash	str	Audit hash (tamper-proof chain)
.raw	dict	Full raw API response

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.