nextriad-agentshield 0.2.0
The $19/mo Permission Layer for AI Agents - Verify, Control, Audit. Block before execution.
Requires Python: >=3.8
Keywords: ai, agents, security, trust, verification, a2a, governance, permissions, mcp, llm, langchain, autogpt, crewai, agent-security, ai-firewall, agent-permissions, agent-audit, blockchain-audit

# AgentShield Python SDK

Security and governance for AI agents.

## Installation

```bash
pip install agentshield
```

## Quick Start

```python
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

```python
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

```python
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

```python
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

```python
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

```python
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

```python
# 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

```python
# 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

```python
# 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

| Type | Description |
|------|-------------|
| `prompt_injection` | Attempts to override system instructions |
| `credential_theft` | Trying to extract API keys/passwords |
| `impersonation` | Pretending to be another agent/user |
| `spam` | Excessive or unwanted communications |
| `data_exfiltration` | Unauthorized data extraction |
| `social_engineering` | Manipulation tactics |
| `coordinated_attack` | Multi-agent attack patterns |
| `other` | Other malicious behavior |

## Async Methods

All methods have async versions:

```python
# 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