Skip to main content

Sovereign AI Security SDK - Protect AI agents from prompt injection and wallet drains

Project description

Kyvern Shield Python SDK

Official Python SDK for Kyvern Shield - Transaction security for AI agents.

Protect your AI agents from prompt injection attacks, wallet drains, and unauthorized transactions.

Installation

pip install kyvern-shield

Quick Start

from kyvern_shield import KyvernShield

# Initialize with your API key
shield = KyvernShield(api_key="sk_live_kyvern_xxxxx")

# Before executing any transaction, verify with Shield
result = shield.analyze(
    intent="Transfer 10 SOL to partner wallet",
    to="7xKXtg2CW87d97TXJSDpbD5jBkheTqA83TZRuJosgAsU",
    amount=10.0,
    reasoning="Payment for API services per invoice #1234"
)

if result.is_blocked:
    print(f"BLOCKED: {result.explanation}")
else:
    print(f"APPROVED (risk: {result.risk_score}/100)")
    # Safe to proceed with transaction

Configuration

Environment Variables

export KYVERN_API_KEY=sk_live_kyvern_xxxxx
export KYVERN_AGENT_ID=my-trading-bot  # Optional default agent ID
export KYVERN_API_URL=https://api.kyvernlabs.com  # Optional, for self-hosted

Client Options

shield = KyvernShield(
    api_key="sk_live_kyvern_xxxxx",  # Or use KYVERN_API_KEY env var
    agent_id="my-trading-bot",        # Default agent ID for requests
    base_url="https://api.kyvernlabs.com",  # API endpoint
    timeout=30.0,                     # Request timeout in seconds
)

Usage

Basic Analysis

from kyvern_shield import KyvernShield

with KyvernShield() as shield:
    result = shield.analyze(
        intent="Swap 5 SOL for USDC",
        to="JUP6LkbZbjS1jKKwapdHNy74zcZ3tLUZoi5QNyVTaV4",
        amount=5.0,
        reasoning="User requested swap via chat interface"
    )

    if result.decision == "allow":
        execute_swap()

Response Properties

result = shield.analyze(...)

# Decision
result.decision      # "allow" or "block"
result.is_blocked    # True if blocked
result.is_allowed    # True if allowed

# Risk Assessment
result.risk_score    # 0-100
result.is_high_risk  # True if risk_score >= 70
result.is_low_risk   # True if risk_score < 30

# Details
result.explanation   # Human-readable explanation
result.request_id    # Unique request ID for debugging

# Layer Results (optional, based on analysis)
result.heuristic_result        # Blacklist/amount checks
result.source_detection_result # Untrusted source detection
result.llm_result             # LLM-based analysis

Async Support

For async/await workflows:

import asyncio
from kyvern_shield import AsyncKyvernShield

async def protect_transaction():
    async with AsyncKyvernShield() as shield:
        result = await shield.analyze(
            intent="Transfer 10 SOL",
            to="7xKXtg2CW87d97TXJSDpbD5jBkheTqA83TZRuJosgAsU",
            amount=10.0,
            reasoning="Automated payment"
        )

        if result.is_allowed:
            await execute_transaction()

asyncio.run(protect_transaction())

Error Handling

from kyvern_shield import (
    KyvernShield,
    KyvernShieldError,
    AuthenticationError,
    ValidationError,
    APIError,
    NetworkError,
)

try:
    result = shield.analyze(...)
except AuthenticationError:
    print("Invalid API key")
except ValidationError as e:
    print(f"Invalid request: {e}")
except NetworkError:
    print("Network error - fail safe, block transaction")
except APIError as e:
    print(f"API error ({e.status_code}): {e}")
except KyvernShieldError as e:
    print(f"Shield error: {e}")

Integration Patterns

Fail-Safe Pattern (Recommended)

Always block transactions if Shield is unreachable:

def protected_transfer(to: str, amount: float, reasoning: str) -> bool:
    try:
        result = shield.analyze(
            intent=f"Transfer {amount} SOL",
            to=to,
            amount=amount,
            reasoning=reasoning,
        )

        if result.is_blocked:
            log_blocked_transaction(result)
            return False

        execute_transfer(to, amount)
        return True

    except KyvernShieldError as e:
        # FAIL SAFE: Don't execute if Shield is unavailable
        log_shield_error(e)
        return False

Human-in-the-Loop Pattern

Require approval for high-risk transactions:

result = shield.analyze(...)

if result.is_blocked:
    reject_transaction()
elif result.is_high_risk:
    queue_for_approval(result)  # Send to human reviewer
else:
    execute_transaction()

Batch Analysis

Analyze multiple transactions efficiently:

import asyncio
from kyvern_shield import AsyncKyvernShield

async def analyze_batch(transactions: list[dict]) -> list:
    async with AsyncKyvernShield() as shield:
        tasks = [
            shield.analyze(
                intent=tx["intent"],
                to=tx["to"],
                amount=tx["amount"],
                reasoning=tx["reasoning"],
            )
            for tx in transactions
        ]
        return await asyncio.gather(*tasks)

API Reference

KyvernShield

Main synchronous client class.

Constructor

KyvernShield(
    api_key: str | None = None,    # API key (or use env var)
    base_url: str | None = None,   # API base URL
    timeout: float = 30.0,         # Request timeout
    agent_id: str | None = None,   # Default agent ID
)

Methods

  • analyze(intent, to, amount, reasoning, *, agent_id=None, function_signature="transfer") - Analyze transaction intent
  • health_check() - Check API health
  • close() - Close HTTP client

AsyncKyvernShield

Async version with identical API, using await for all methods.

AnalysisResult

Response from analyze() method.

Property Type Description
request_id str Unique request identifier
decision "allow" | "block" Shield decision
risk_score int Risk score 0-100
explanation str Human-readable explanation
is_blocked bool True if decision is "block"
is_allowed bool True if decision is "allow"
is_high_risk bool True if risk_score >= 70
is_low_risk bool True if risk_score < 30

Requirements

  • Python 3.9+
  • httpx >= 0.25.0
  • pydantic >= 2.0.0

License

MIT License - see LICENSE for details.

Support

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

kyvern_shield-0.1.0.tar.gz (10.9 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

kyvern_shield-0.1.0-py3-none-any.whl (9.2 kB view details)

Uploaded Python 3

File details

Details for the file kyvern_shield-0.1.0.tar.gz.

File metadata

  • Download URL: kyvern_shield-0.1.0.tar.gz
  • Upload date:
  • Size: 10.9 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.9.6

File hashes

Hashes for kyvern_shield-0.1.0.tar.gz
Algorithm Hash digest
SHA256 4e56de445ea7b66fda1d7169ebe8fd8a30de8c69752124fc15032e4177ff9cb3
MD5 912b4f653aea29dffdce68a00c9be6f3
BLAKE2b-256 fcdcb80675f79cb9c6c2c3652d879c698c261e5eb80a30c72d71bec2ae744399

See more details on using hashes here.

File details

Details for the file kyvern_shield-0.1.0-py3-none-any.whl.

File metadata

  • Download URL: kyvern_shield-0.1.0-py3-none-any.whl
  • Upload date:
  • Size: 9.2 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.9.6

File hashes

Hashes for kyvern_shield-0.1.0-py3-none-any.whl
Algorithm Hash digest
SHA256 d949907f80e50308027758fa7e922b8e0a6053572093588d84100d64044d48d2
MD5 25860162ffb1c515d802a67087a07e94
BLAKE2b-256 b171a0dc5e126bd5ab19691ab20948cb30b43ad7f862f9de300005b359a2e17a

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page