AgentShield Python SDK

Security and governance for AI agents.

Installation

pip install agentshield

Quick Start

from agentshield import AgentShield

# Initialize with your API key
shield = AgentShield(api_key="as_live_your_key_here")

# Option 1: Decorator (recommended)
@shield.protect(scope="email.send")
def send_email(to: str, subject: str, body: str):
    # Your email sending logic here
    # This will only execute if AgentShield approves the action
    pass

# Option 2: Manual verification
result = shield.verify(scope="files.read")
if result["allowed"]:
    read_files()
else:
    print(f"Blocked: {result['reason']}")

Async Support

import asyncio
from agentshield import AgentShield

shield = AgentShield(api_key="as_live_xxx")

# Async decorator
@shield.protect(scope="api.call")
async def make_api_call():
    # Your async logic here
    pass

# Async context manager
async def main():
    async with shield.action(scope="database.write") as ctx:
        await write_to_database()
        ctx.report_success({"rows_affected": 10})

asyncio.run(main())

Handling Approvals

Some actions require human approval. You can handle this in two ways:

Option 1: Auto-wait for approval

shield = AgentShield(
    api_key="as_live_xxx",
    auto_wait_approval=True,  # Wait up to 5 minutes for approval
    approval_timeout=300
)

@shield.protect(scope="payment.process")
def process_payment(amount: float):
    # This will block until a human approves or rejects
    # the action in the AgentShield dashboard
    pass

Option 2: Handle approval manually

from agentshield import AgentShield, ApprovalRequiredError

shield = AgentShield(api_key="as_live_xxx")

try:
    result = shield.verify(scope="email.send")
    if result["allowed"]:
        send_email()
    elif result.get("pending_approval"):
        approval_id = result["approval_id"]
        print(f"Waiting for approval: {approval_id}")
        # Check dashboard or wait...
        approved = await shield.wait_for_approval(approval_id, timeout=60)
        if approved:
            send_email()
except ApprovalRequiredError as e:
    print(f"Need approval: {e.approval_id}")

Error Handling

from agentshield import (
    AgentShield,
    NotAuthorizedError,
    RateLimitError,
    ApprovalRequiredError
)

shield = AgentShield(api_key="as_live_xxx")

try:
    @shield.protect(scope="code.execute")
    def run_code():
        exec(user_code)
    
    run_code()

except NotAuthorizedError as e:
    print(f"Not authorized: {e.scope} - {e.reason}")

except RateLimitError as e:
    print(f"Rate limited on {e.scope}")
    if e.retry_after:
        print(f"Retry after {e.retry_after} seconds")

except ApprovalRequiredError as e:
    print(f"Needs approval: {e.approval_id}")

Configuration

shield = AgentShield(
    api_key="as_live_xxx",
    base_url="https://api.agentshield.io",  # Custom API URL
    timeout=30.0,  # Request timeout in seconds
    auto_wait_approval=False,  # Auto-wait for approvals
    approval_timeout=300  # Max time to wait for approval
)

Threat Intelligence

AgentShield includes built-in threat detection and reporting.

Detect Threats in Content

# Analyze user input for prompt injection, social engineering, etc.
result = shield.detect_threats(user_input)

if result["detected"]:
    print(f"⚠️ Threat detected: {result['threat_type']}")
    print(f"Confidence: {result['confidence']}")
    print(f"Patterns: {result['patterns_matched']}")

Check Blacklist

# Check if an agent is known to be malicious
result = shield.check_blacklist("agent_xyz")

if result["is_blacklisted"]:
    print(f"🚫 Agent is blacklisted!")
    print(f"Threat count: {result['threat_count']}")
    for threat in result["threats"]:
        print(f"  - {threat['type']}: {threat['evidence']}")

Report Threats

# Report a malicious agent to the community blacklist
shield.report_threat(
    agent_id="malicious_agent_123",
    threat_type="prompt_injection",  # or: credential_theft, impersonation, spam, etc.
    severity="high",  # low, medium, high, critical
    evidence="Attempted to override system prompt with 'ignore all instructions'",
    source_ip="192.168.1.1",  # optional
    context={"conversation_id": "abc123"}  # optional
)

Threat Types

Async Methods

All methods have async versions:

# Sync
result = shield.verify(scope="email.send")
result = shield.check_blacklist("agent_id")
result = shield.detect_threats("content")
shield.report_threat(agent_id="...", threat_type="...")

# Async
result = await shield.verify_async(scope="email.send")
result = await shield.check_blacklist_async("agent_id")
result = await shield.detect_threats_async("content")
await shield.report_threat_async(agent_id="...", threat_type="...")

License

MIT