Python SDK

git clone https://github.com/shieldagent-io/shieldagent.git
pip install -e ./shieldagent/backend/packages/sdk-python

from shieldagent import ShieldAgentClient

client = ShieldAgentClient(
    api_key="{your-api-key}",
    base_url="https://api.shieldagent.io",
)

# List agents for a tenant
agents = client.agents.list(tenant_id="ten_abc123")
print(agents)

# Register a new agent
agent = client.agents.create(
    tenant_id="ten_abc123",
    name="my-assistant",
    description="Python-based coding agent",
)
print(agent.agent_key)  # sa_live_...

# Fetch recent audit events
events = client.audit.list(
    tenant_id="ten_abc123",
    limit=50,
    order="desc",
)
for event in events:
    print(event.tool, event.decision, event.risk_score)

import asyncio
from shieldagent import ShieldAgentClient, ForbiddenError, RateLimitError, BadGatewayError

async def main():
    async with ShieldAgentClient(
        api_key="{your-agent-key}",
        base_url="https://proxy.shieldagent.io",
    ) as client:

        # ── POST through the proxy ────────────────────────────────────────
        charge = await client.api_proxy(
            "stripe", "POST", "/charges",
            json={"amount": 2000, "currency": "usd", "source": "tok_visa"},
        )

        # ── GET a specific resource ───────────────────────────────────────
        customer = await client.api_proxy(
            "stripe", "GET", "/customers/cus_abc123",
        )

        # ── Send a Slack message ──────────────────────────────────────────
        await client.api_proxy(
            "slack", "POST", "/chat.postMessage",
            json={"channel": "#alerts", "text": "Anomaly detected"},
        )

        # ── Call an internal microservice ─────────────────────────────────
        result = await client.api_proxy(
            "order-service", "POST", "/orders",
            json={"customer_id": "cust_42", "items": [{"sku": "ABC", "qty": 1}]},
        )

        # ── Error handling ────────────────────────────────────────────────
        try:
            await client.api_proxy("stripe", "DELETE", "/charges/ch_123")
        except ForbiddenError as e:
            print("Blocked by policy:", e.code, e.message)
        except RateLimitError as e:
            print("Rate limited. Retry after:", e.retry_after)
        except BadGatewayError as e:
            print("Upstream error:", e.message)

asyncio.run(main())

from shieldagent import SyncShieldAgentClient

with SyncShieldAgentClient(api_key="...", base_url="...") as client:
    charge = client.api_proxy(
        "stripe", "POST", "/charges",
        json={"amount": 2000, "currency": "usd", "source": "tok_visa"},
    )

import asyncio
from shieldagent import ShieldAgentClient

async def main():
    async with ShieldAgentClient(
        api_key="{your-agent-key}",
        base_url="https://api.shieldagent.io",
        proxy_url="https://proxy.shieldagent.io",
    ) as client:

        # ── 1. Scan before executing ──────────────────────────────────────
        verdict = await client.scan(
            tenant_id="ten_abc123",
            tool_name="read_file",
            params={"path": "/etc/passwd"},
            agent_id="agt_xyz789",       # optional — improves policy matching
            include_findings=True,        # optional — request detailed findings
        )

        # IMPORTANT: YOU must enforce the verdict. ShieldAgent does not block.
        if verdict.action == "block":
            raise RuntimeError(f"Tool call blocked: {verdict.reason}")
        if verdict.action == "human_review":
            raise RuntimeError(
                f"Tool call requires human review (review_id: {verdict.review_id})"
            )

        # ── 2. Execute the tool ───────────────────────────────────────────
        file_content = read_file("/etc/passwd")

        # ── 3. Confirm execution to close the audit loop ──────────────────
        await client.confirm_execution(
            "ten_abc123", verdict.audit_event_id, executed=True
        )

        # ── Verdict fields ────────────────────────────────────────────────
        # verdict.action         — 'allow' | 'block' | 'human_review'
        # verdict.reason         — human-readable explanation
        # verdict.risk_score     — 0–100 risk score at scan time
        # verdict.audit_event_id — use with confirm_execution()
        # verdict.review_id      — set when action is 'human_review', else None
        # verdict.findings       — list of findings (when include_findings=True)

asyncio.run(main())

from shieldagent import SyncShieldAgentClient

with SyncShieldAgentClient(api_key="...", base_url="...", proxy_url="...") as client:
    verdict = client.scan("ten_abc123", "read_file", {"path": "/etc/passwd"})
    if verdict.action == "block":
        raise RuntimeError(verdict.reason)
    # ... execute tool ...
    client.confirm_execution("ten_abc123", verdict.audit_event_id, executed=True)

async with ShieldAgentClient(api_key="...", base_url="...") as client:
    # Apply a built-in template to the tenant
    result = await client.policy_templates.apply("ten_abc123", "pii-protection")
    print(f"Created {result.policies_created} policies from '{result.template_name}'")

    # Block DELETE on the Stripe endpoint for a specific agent
    await client.policies.create(
        "ten_abc123",
        agent_id=agent.id,
        tool_name="rest:stripe",
        action="deny",
        conditions={"method": "DELETE"},
    )

    # List active policies for an agent
    policies = await client.policies.list("ten_abc123", agent_id=agent.id)
    for p in policies.data:
        print(p.tool_name, p.action)

async with ShieldAgentClient(api_key="...", base_url="...") as client:
    # Get current risk scores
    scores = await client.risk.list("ten_abc123")
    high_risk = [s for s in scores.data if s.risk_tier in ("high", "critical")]
    print("High-risk agents:", [s.agent_name for s in high_risk])

    # Drill into a specific agent's risk trend
    detail = await client.risk.get_detail("ten_abc123", agent.id)
    print(f"Risk: {detail.score} ({detail.trend}) — delta 24h: {detail.delta_vs_24h}")

    # Batch anomaly analysis over the last 24 hours
    anomalies = await client.anomalies.analyze_batch(
        "ten_abc123", agent_id=agent.id, window_hours=24
    )
    if anomalies.anomalies_detected:
        for a in anomalies.results:
            print(f"Anomaly [{a.anomaly_type}] score={a.anomaly_score:.2f} — {a.explanation}")

async with ShieldAgentClient(api_key="...", base_url="...") as client:
    # Open an incident for a detected threat
    incident = await client.incidents.create(
        "ten_abc123",
        agent_id=agent.id,
        title="Prompt injection detected in session ses_xyz",
        severity="critical",
        type="injection",
        data={"session_id": "ses_xyz"},
    )

    # Move to investigating
    await client.incidents.update(incident.id, status="investigating")

    # Process the human-review queue
    pending = await client.reviews.list("ten_abc123", status="pending")
    for review in pending.data:
        decision = "denied" if "stripe" in review.tool_name else "approved"
        await client.reviews.decide(
            review.id, decision=decision, comment="Reviewed by on-call"
        )

from shieldagent import ShieldAgentClient
import asyncio

async def main():
    async with ShieldAgentClient(api_key="...") as client:
        agents = await client.agents.list(tenant_id="ten_abc123")
        print(agents)

asyncio.run(main())

from shieldagent import ShieldAgentError

try:
    client.agents.create(tenant_id="bad-id", name="x")
except ShieldAgentError as e:
    print(e.status, e.code, e.message)