guardian-agent-sdk 1.0.2

Guardian Agent Python SDK for AI Security - Real-time governance for autonomous AI agents

guardian-agent-sdk

Real-time governance for autonomous AI agents — evaluate tool calls (allow / block / ask) before side effects run, poll for human approval, and manage policies and agents against a Guardian Agent backend.

Install from PyPI: pypi.org/project/guardian-agent-sdk

---

Why Guardian?

LLM agents can execute SQL, send email, call APIs, and modify production systems. The Guardian SDK sits at the tool boundary: every call is checked against your policies before it runs.

• Block destructive or out-of-policy actions immediately
• Ask for human review on high-risk operations (HITL)
• Allow routine operations with a full audit trail

---

Features

Area	Capability
Policy evaluation	client.events.check() — server-side deterministic rules + optional LLM judge
Human-in-the-loop	client.events.wait_for_approval() — poll events and review tasks
Resilience	Retries with jitter, HTTP connection pooling, circuit breaker
Fail-safe	fail_closed / fail_open / local_scan when the backend is unavailable
Observability	W3C traceparent propagation, optional structured JSON logging
API resources	Events, policies, agents, webhooks, metrics, audit
Runtimes	Async (GuardianClient) and sync (SyncGuardianClient)

Requirements: Python 3.11+

---

Installation

pip install guardian-agent-sdk

You need a running Guardian backend and tenant API credentials. See the Quick Start guide in the main repository.

For WebSocket live feeds (client.events.stream()), websockets is included as a dependency.

---

Quick start

import asyncio
from guardian_sdk import GuardianClient
from guardian_sdk.config import SDKConfig

async def main():
    client = GuardianClient(
        config=SDKConfig(
            api_key="gdn_live_...",       # operator or per-agent key
            tenant_id="your_org_id",      # X-Tenant-ID
            agent_id="agent_dev_01",
            base_url="http://127.0.0.1:8000",  # host only — see below
        )
    )
    try:
        result = await client.events.check(
            tool_name="execute_sql",
            tool_input={"query": "SELECT id FROM users LIMIT 10"},
        )
        print(result.decision, result.event_id, result.reason)

        if result.decision == "ask":
            approval = await client.events.wait_for_approval(
                result.event_id,
                poll_interval=2,
                max_wait_seconds=300,
            )
            print("Approved:", approval.get("decision_token"))
    finally:
        await client.close()

asyncio.run(main())

Sync client:

from guardian_sdk import SyncGuardianClient

with SyncGuardianClient(api_key="...", tenant_id="...") as client:
    result = client.events.check("execute_sql", {"query": "SELECT 1"})
    print(result.decision)

From environment variables:

from guardian_sdk import GuardianClient

client = GuardianClient.from_env()

export GUARDIAN_API_KEY="gdn_live_..."
export GUARDIAN_TENANT_ID="your_org_id"
export GUARDIAN_BASE_URL="http://127.0.0.1:8000"
export GUARDIAN_AGENT_ID="agent_dev_01"

---

Configuration essentials

base_url — host only

Set base_url to the API host, not /api/v1:

Correct	Incorrect
http://127.0.0.1:8000	http://127.0.0.1:8000/api/v1

The SDK appends paths like /v1/events/check. Including /api/v1 in base_url causes double-prefixed URLs and 404 errors.

Production default: https://api.guardian.app

Authentication

Operator and agent runtimes use:

Header	Purpose
X-Tenant-ID	Organization id (required)
X-Guardian-API-Key	API key (required)

The SDK sets these automatically from SDKConfig. Dashboard user sessions may use Authorization: Bearer (JWT) — that path is separate from agent events.check integration.

Enterprise options (optional)

export GUARDIAN_FAIL_SAFE_MODE="fail_closed"   # fail_closed | fail_open | local_scan
export GUARDIAN_CIRCUIT_BREAKER="true"
export GUARDIAN_STRUCTURED_LOG="true"
export GUARDIAN_TRACING="true"
export GUARDIAN_MAX_RETRIES="3"

from guardian_sdk.config import SDKConfig, FailSafeMode

config = SDKConfig(
    fail_safe_mode=FailSafeMode.FAIL_CLOSED,
    circuit_breaker_enabled=True,
    structured_logging=True,
    enable_tracing=True,
)

---

Core API

Operation	Method
Policy check	POST /v1/events/check
Get event	GET /v1/events/{event_id}
Review tasks	GET /v1/review-tasks
Approve / deny	POST /v1/events/{event_id}/approve or /deny

Register agents with the operator key:

reg = await client.agents.register(name="prod-worker-1", agent_type="assistant")
print(reg.agent.id, reg.api_key)  # api_key shown once — store securely

---

PolicyDecisionResponse

events.check() returns a typed Pydantic model:

• decision, event_id, reason — core outcome
• policy_id, rule_name, matched_pattern — matched rule
• decision_token, execution_id — post-approval execution binding
• pending_review — True when decision is ask
• why_trace, preflight, match_kind, decision_id — enterprise audit metadata

Use result.preflight.code for stable branching on preflight errors (e.g. unknown agent).

---

Resilience & fail-safe

Circuit breaker — trips on repeated 5xx / network errors; fail-fast while open:

print(client.circuit_breaker_snapshot.state)

Fail-safe modes (when backend or circuit is unavailable during events.check):

Mode	Behavior
fail_closed (default)	Return block
fail_open	Return allow with fail_safe metadata
local_scan	Offline regex pre-scan (SQL injection patterns, path traversal, etc.)

Retries — transient 429/503 with exponential backoff and Retry-After support.

---

WebSocket live feed

async for msg in client.events.stream():
    print(msg.decision, msg.tool_name, msg.event_id)

Connects to /ws/v1/events/stream. Auth handshake:

{"type": "auth", "tenant_id": "<org>", "api_key": "<key>"}

Reply to server ping with {"type": "pong"}.

---

Examples & documentation

Resource	Link
Fintech demo (SQL block + email HITL)	examples/fintech_demo.py
LangGraph tool guard	examples/langgraph_tool_guard.py
Full stack quick start	docs/QUICK_START.md
HTTP / curl integration	docs/HTTP_AGENT_INTEGRATION.md
LLM Judge (backend)	docs/LLM_JUDGE_GUIDE.md

---

Testing without a backend

from guardian_sdk import GuardianClient

client = GuardianClient.create_mock_client(
    responses=[{"decision": "allow", "event_id": "evt_mock", "reason": "ok"}],
)
result = await client.events.check("echo_tool", {"line": "hello"})

---

Links

• PyPI: https://pypi.org/project/guardian-agent-sdk/
• Source: https://github.com/natnaeldejenekebede/guardian-agent
• Issues: https://github.com/natnaeldejenekebede/guardian-agent/issues
• Documentation: https://docs.guardian-agent.com

---

License

MIT — see LICENSE in the repository.