capiscio-sdk 2.5.0

Runtime security middleware for A2A agents

CapiscIO Agent Guard

Runtime identity and trust enforcement for A2A agents.

Agent Guard is the CapiscIO runtime SDK for A2A Protocol agents. It provides identity management, request signing, badge verification, and trust enforcement with zero configuration.

🚀 The 60-Second Upgrade

Turn any FastAPI application into a Verified A2A Agent in 3 lines of code.

from fastapi import FastAPI
from capiscio_sdk.simple_guard import SimpleGuard
from capiscio_sdk.integrations.fastapi import CapiscioMiddleware

# 1. Initialize Guard (Auto-generates keys in dev_mode)
guard = SimpleGuard(dev_mode=True)

app = FastAPI()

# 2. Add Enforcement Middleware
app.add_middleware(CapiscioMiddleware, guard=guard)

@app.post("/agent/task")
async def handle_task(request: Request):
    # 🔒 Only reachable if Identity + Integrity are verified
    caller = request.state.agent_id
    return {"status": "accepted", "verified_caller": caller}

Configurable Security Modes

Control enforcement behavior via environment variables:

from capiscio_sdk.config import SecurityConfig
from capiscio_sdk.integrations.fastapi import CapiscioMiddleware

# Load config from environment variables
config = SecurityConfig.from_env()

# Or use presets
config = SecurityConfig.production()  # Balanced production defaults (block on verified failures)
config = SecurityConfig.development() # Monitor mode (log but allow)

app.add_middleware(CapiscioMiddleware, guard=guard, config=config)

Environment Variables:

Variable	Description	Default
CAPISCIO_REQUIRE_SIGNATURES	Require badge on requests	false
CAPISCIO_FAIL_MODE	block, monitor, or log	block
CAPISCIO_RATE_LIMIT_RPM	Rate limit (requests/min)	60

Middleware Observability (Auto-Events)

Enable automatic event emission from the middleware to get visibility into request patterns, verification outcomes, and latency — no manual instrumentation required.

from capiscio_sdk.events import EventEmitter
from capiscio_sdk.integrations.fastapi import CapiscioMiddleware

emitter = EventEmitter(agent_id="...", api_key="...", registry_url="...")
app.add_middleware(CapiscioMiddleware, guard=guard, emitter=emitter)
# Events flow automatically — no other code changes needed

The middleware emits these events when an emitter is provided:

Event	When	Key Fields
request.received	Every inbound request	method, path
verification.success	Badge verified	method, path, caller_did, duration_ms
verification.failed	Badge missing/invalid	method, path, reason, duration_ms
request.completed	Response sent	method, path, status_code, duration_ms, caller_did

Privacy note: Auto-events are strictly opt-in. No telemetry is sent unless you explicitly pass an emitter. Excluded paths emit no events.

🛡️ What You Get (Out of the Box)

1. Zero-Config Identity:

   • Auto-generates Ed25519 keys and agent-card.json on first run.
   • No manual key management required for development.

2. Payload Integrity:

   • Enforces SHA-256 Body Hash (bh) verification.
   • Blocks tampered payloads instantly (returns 403 Forbidden).

3. Replay Protection:

   • Enforces strict 60-second token expiration (exp).
   • Prevents replay attacks and ensures freshness.

4. Performance Telemetry:

   • Adds <1ms overhead.
   • Includes Server-Timing headers for transparent monitoring.

Installation

pip install capiscio-sdk

🔌 CapiscIO.connect() - "Let's Encrypt" Style Setup

The fastest way to get a production-ready agent identity:

from capiscio_sdk import CapiscIO

# One-liner to get a fully configured agent
agent = CapiscIO.connect(api_key="sk_live_...")

# Agent is now ready
print(agent.did)    # did:web:registry.capisc.io:agents:...
print(agent.badge)  # Current badge (auto-renewed)
print(agent.name)   # Agent name

Connect Parameters

Parameter	Description	Default
api_key	Your CapiscIO API key	Required
name	Agent name for lookup/creation	Auto-generated
agent_id	Specific agent UUID	Auto-discovered
server_url	Registry URL	https://registry.capisc.io
auto_badge	Request badge automatically	True
dev_mode	Use self-signed badges (Level 0)	False

Agent Discovery & Recovery

The SDK uses a smart discovery flow to find or create your agent:

1. Local keys check - Scans ~/.capiscio/keys/*/ for any agent subdirectory with valid keys
2. DID-based verification - Verifies local DID matches server's DID before reusing identity
3. Name-based lookup - Falls back to searching by name if no local keys found
4. Auto-create - Creates new agent if not found

# First run: creates "my-research-agent" + generates keys
agent = CapiscIO.connect(api_key="...", name="my-research-agent")

# Second run: finds agent by existing keys (even if name changed in dashboard)
agent = CapiscIO.connect(api_key="...", name="my-research-agent")

Key directory structure:

~/.capiscio/keys/
└── {agent-uuid}/
    ├── private.jwk      # Ed25519 private key
    ├── public.jwk       # Public key (contains DID in 'kid' field)
    └── agent-card.json  # Agent metadata

Recovery behavior: If the SDK finds existing keys but the agent's DID isn't registered with the server, it will automatically complete registration. This handles crash recovery scenarios gracefully.

Using Environment Variables

from capiscio_sdk import CapiscIO

# Reads from CAPISCIO_API_KEY, CAPISCIO_AGENT_NAME, etc.
agent = CapiscIO.from_env()

Environment Variables:

Variable	Required	Description
CAPISCIO_API_KEY	Yes	Your API key
CAPISCIO_AGENT_NAME	No	Agent name for lookup/creation
CAPISCIO_AGENT_ID	No	Specific agent UUID
CAPISCIO_SERVER_URL	No	Registry URL (default: production)
CAPISCIO_DEV_MODE	No	Enable dev mode (true/false)
CAPISCIO_AGENT_PRIVATE_KEY_JWK	No	JSON-encoded Ed25519 private JWK for ephemeral environments

Deploying to Containers / Serverless

In ephemeral environments (Docker, Lambda, Cloud Run) the local ~/.capiscio/ directory doesn't survive restarts. On first run the SDK generates a keypair and logs a capture hint:

╔══════════════════════════════════════════════════════════════════╗
║  New agent identity generated — save key for persistence         ║
╚══════════════════════════════════════════════════════════════════╝

  Add to your secrets manager / .env:

    CAPISCIO_AGENT_PRIVATE_KEY_JWK='{"kty":"OKP","crv":"Ed25519","d":"...","x":"...","kid":"did:key:z6Mk..."}'

Copy that value into your secrets manager and set it as an environment variable. On subsequent starts the SDK recovers the same DID without generating a new identity.

Key resolution priority: env var → local file → generate new.

# docker-compose.yml
services:
  my-agent:
    environment:
      CAPISCIO_API_KEY: "sk_live_..."
      CAPISCIO_AGENT_PRIVATE_KEY_JWK: "${AGENT_KEY_JWK}"  # from secrets

See the Configuration Guide for full deployment examples.

🎯 Agent Card Validation with CoreValidator

The SDK includes a Go core-backed validator for Agent Card validation. This ensures consistent validation behavior across all CapiscIO SDKs (Python, Node.js, etc.).

Quick Validation

from capiscio_sdk.validators import validate_agent_card

# One-shot validation
result = validate_agent_card({
    "name": "My Agent",
    "url": "https://myagent.example.com",
    "version": "1.0.0",
    "skills": [{"id": "chat", "name": "Chat", "description": "Chat skill"}]
})

print(f"Valid: {result.success}")
print(f"Compliance Score: {result.compliance.total}/100 ({result.compliance.rating})")
print(f"Trust Score: {result.trust.total}/100 ({result.trust.rating})")

Multi-Dimensional Scoring

CoreValidator returns rich multi-dimensional scores:

from capiscio_sdk.validators import CoreValidator

with CoreValidator() as validator:
    result = validator.validate_agent_card(card)
    
    # 📊 Compliance Score (0-100)
    print(f"Compliance: {result.compliance.total}")
    print(f"  - Core Fields: {result.compliance.breakdown.core_fields.score}")
    print(f"  - Skills Quality: {result.compliance.breakdown.skills_quality.score}")
    print(f"  - Format Compliance: {result.compliance.breakdown.format_compliance.score}")
    print(f"  - Data Quality: {result.compliance.breakdown.data_quality.score}")
    
    # 🔒 Trust Score (0-100)
    print(f"Trust: {result.trust.total}")
    print(f"  - Signatures: {result.trust.breakdown.signatures.score}")
    print(f"  - Provider: {result.trust.breakdown.provider.score}")
    print(f"  - Security: {result.trust.breakdown.security.score}")
    print(f"  - Documentation: {result.trust.breakdown.documentation.score}")
    
    # 📡 Availability Score (when tested)
    if result.availability.tested:
        print(f"Availability: {result.availability.total}")

Score Ratings

Score Range	Compliance Rating	Trust Rating
90-100	A+	Verified
80-89	A	High
70-79	B	Good
60-69	C	Moderate
50-59	D	Low
0-49	F	Untrusted

Async Fetch and Validate

async def check_remote_agent(agent_url: str):
    with CoreValidator() as validator:
        result = await validator.fetch_and_validate(agent_url)
        return result.success and result.compliance.total >= 80

Migration from Legacy Validators

# ❌ Deprecated (will be removed in v1.0.0)
from capiscio_sdk.validators import AgentCardValidator
validator = AgentCardValidator()  # Shows deprecation warning

# ✅ Recommended
from capiscio_sdk.validators import CoreValidator, validate_agent_card
result = validate_agent_card(card)  # Uses Go core

🔏 Trust Badge Verification (RFC-002)

CapiscIO Trust Badges provide cryptographic proof of agent identity verification. The SDK supports verifying badges issued by the CapiscIO registry or self-signed badges for development.

Trust Levels

Level	Name	Description
0	Self-Signed (SS)	No external validation, did:key issuer
1	Domain Validated (DV)	Domain ownership verified
2	Organization Validated (OV)	Organization identity verified
3	Extended Validated (EV)	Highest level of identity verification
4	Community Vouched (CV)	Verified with peer attestations

Verify a Trust Badge

from capiscio_sdk._rpc.client import CapiscioRPCClient

# Connect to gRPC server (auto-starts if needed)
client = CapiscioRPCClient()
client.connect()

# Badge token (JWS format)
badge_token = "eyJhbGciOiJFZERTQSJ9.eyJqdGkiOi..."

# Verify badge (production - rejects self-signed)
valid, claims, warnings, error = client.badge.verify_badge_with_options(
    badge_token,
    accept_self_signed=False  # Default: reject self-signed badges
)

if valid:
    print(f"✅ Badge verified!")
    print(f"   Issuer: {claims['iss']}")
    print(f"   Subject: {claims['sub']}")
    print(f"   Trust Level: {claims['trust_level']}")  # "0", "1", "2", "3", or "4"
    print(f"   Expires: {claims['exp']}")
else:
    print(f"❌ Verification failed: {error}")

client.close()

Accept Self-Signed Badges (Development)

For development/testing, you can accept self-signed (Level 0) badges:

# Development mode - accept self-signed badges
valid, claims, warnings, error = client.badge.verify_badge_with_options(
    badge_token,
    accept_self_signed=True  # ⚠️ Only for development!
)

if valid and claims['trust_level'] == '0':
    print("⚠️ Self-signed badge - not suitable for production")

Verification Options

# Full verification options
valid, claims, warnings, error = client.badge.verify_badge_with_options(
    token=badge_token,
    accept_self_signed=False,      # Reject Level 0 badges
    trusted_issuers=["https://registry.capisc.io"],  # Allowlist of trusted CAs
    audience="my-service",         # Expected audience claim
    skip_revocation=False,         # Check revocation status
    skip_agent_status=False        # Check agent is not disabled
)

Badge Claims Structure

# Example claims from a verified badge
claims = {
    "jti": "550e8400-e29b-41d4-a716-446655440000",  # Unique badge ID
    "iss": "https://registry.capisc.io",            # Issuer URL
    "sub": "did:web:registry.capisc.io:agents:abc123",  # Agent DID
    "iat": 1702234567,                              # Issued at (Unix timestamp)
    "exp": 1702320967,                              # Expires at (Unix timestamp)
    "trust_level": "2",                             # "0"-"4"
    "aud": ["my-service"],                          # Audience (optional)
}

🔌 gRPC SDK Integration (All 7 Services)

The SDK provides a comprehensive gRPC client that wraps all capiscio-core functionality. The client auto-starts the local gRPC server if needed.

Connection Modes

from capiscio_sdk._rpc.client import CapiscioRPCClient

# Auto-start mode (recommended for local development)
client = CapiscioRPCClient()  # Auto-starts capiscio-core gRPC server
client.connect()

# Connect to existing Unix socket
client = CapiscioRPCClient(address="unix:///tmp/capiscio.sock", auto_start=False)
client.connect()

# Connect to remote TCP server
client = CapiscioRPCClient(address="localhost:50051", auto_start=False)
client.connect()

# Context manager (auto-cleanup)
with CapiscioRPCClient() as client:
    # Use client...
    pass  # Automatically disconnects

1. BadgeService - Trust Badge Operations

import time

with CapiscioRPCClient() as client:
    # Sign a badge
    token, claims = client.badge.sign_badge(
        claims={
            "jti": "550e8400-e29b-41d4-a716-446655440000",
            "iss": "https://registry.capisc.io",
            "sub": "did:web:registry.capisc.io:agents:my-agent",
            "iat": int(time.time()),
            "exp": int(time.time()) + 300,
            "trust_level": "2"
        },
        private_key_jwk='{"kty":"OKP",...}',
        key_id="key-1"
    )

    # Verify with full options
    valid, claims, warnings, error = client.badge.verify_badge_with_options(
        token,
        accept_self_signed=True,
        trusted_issuers=["https://registry.capisc.io"],
        audience="my-service"
    )

    # Parse without verification
    claims, error = client.badge.parse_badge(token)

    # Request CA-signed badge
    token, error = client.badge.request_badge(
        agent_id="my-agent-123",
        api_key="capi_key_...",
        ca_url="https://registry.capisc.io"
    )

    # Start badge keeper (auto-renewal)
    for event in client.badge.start_keeper(
        private_key_jwk='{"kty":"OKP",...}',
        subject="did:web:registry.capisc.io:agents:my-agent",
        ca_url="https://registry.capisc.io",
        ttl_seconds=300,
        renew_before_seconds=60
    ):
        if event.event_type == "renewed":
            print(f"Badge renewed: {event.badge_token}")

2. DIDService - DID Parsing

with CapiscioRPCClient() as client:
    # Parse did:web identifier
    did_info, error = client.did.parse("did:web:registry.capisc.io:agents:my-agent")

    if did_info:
        print(f"Method: {did_info['method']}")          # "web"
        print(f"Domain: {did_info['domain']}")          # "registry.capisc.io"
        print(f"Path: {did_info['path']}")              # "agents/my-agent"
        print(f"Document URL: {did_info['document_url']}")  # "https://registry.capisc.io/agents/my-agent/did.json"

3. TrustStoreService - Manage Trusted CA Keys

with CapiscioRPCClient() as client:
    # Add trusted CA key
    kid, error = client.trust.add_key(
        did="did:web:registry.capisc.io",
        public_key=b'{"kty":"OKP",...}',
        format="JWK"
    )

    print(f"Added key: {kid}")

4. RevocationService - Check Revocation Status

with CapiscioRPCClient() as client:
    # Check if badge is revoked
    is_revoked = client.revocation.is_revoked("badge-jti-12345")

    if is_revoked:
        print("⚠️ Badge has been revoked!")

5. ScoringService - Agent Card Scoring

with CapiscioRPCClient() as client:
    # Score an agent card
    result, error = client.scoring.score_agent_card(agent_card_json)

    if result:
        print(f"Overall Score: {result['overall_score']}/100")
        print(f"Compliance: {result['compliance_score']}/100")
        print(f"Trust: {result['trust_score']}/100")
        print(f"Availability: {result['availability_score']}/100")

    # Validate a specific rule
    rule_result, error = client.scoring.validate_rule("rule-001", agent_card_json)

    # List available rule sets
    rule_sets, error = client.scoring.list_rule_sets()

    # Get specific rule set
    rule_set, error = client.scoring.get_rule_set("default")

    # Aggregate multiple scores
    aggregated, error = client.scoring.aggregate_scores(
        results=[
            {"overall_score": 85},
            {"overall_score": 90},
            {"overall_score": 88}
        ],
        method="average"  # or "min", "max"
    )

6. SimpleGuardService - JWS Signing & Verification

with CapiscioRPCClient() as client:
    # Sign a payload
    signature, error = client.simpleguard.sign(
        payload=b"important message",
        key_id="my-key-1"
    )

    # Verify a signature
    valid, payload, error = client.simpleguard.verify(
        signature=signature,
        expected_payload=b"important message",
        public_key_jwk='{"kty":"OKP",...}'
    )

    # Sign with attached payload (JWS Compact)
    jws, error = client.simpleguard.sign_attached(
        payload=b"message",
        key_id="my-key-1"
    )

    # Verify attached signature
    valid, payload, error = client.simpleguard.verify_attached(
        jws=jws,
        public_key_jwk='{"kty":"OKP",...}'
    )

    # Get key information
    key_info, error = client.simpleguard.get_key_info("my-key-1")

7. RegistryService - Fetch Agent Cards

with CapiscioRPCClient() as client:
    # Get agent card by DID
    agent_card, error = client.registry.get_agent(
        did="did:web:registry.capisc.io:agents:my-agent"
    )

    if agent_card:
        print(f"Agent Name: {agent_card['name']}")
        print(f"URL: {agent_card['url']}")

Process Manager

The SDK includes automatic process management for the capiscio-core gRPC server:

from capiscio_sdk._rpc.process import get_process_manager

# Get singleton process manager
pm = get_process_manager()

# Ensure server is running (auto-starts if needed)
address = pm.ensure_running(timeout=10.0)
print(f"gRPC server running at: {address}")

# Manually start server
pm.start()

# Stop server
pm.stop()

# Check if running
if pm.is_running():
    print("Server is running")

Auto-Start Behavior:

• ✅ Automatically downloads capiscio-core binary if not found
  • Downloads from GitHub releases (capiscio/capiscio-core)
  • Supports macOS (arm64/x86_64), Linux (arm64/x86_64), and Windows
  • Caches binary in ~/.capiscio/bin/ for reuse
  • Sets executable permissions automatically on Unix-like systems
• ✅ Starts on Unix socket by default (~/.capiscio/rpc.sock)
• ✅ Handles server crashes and restarts
• ✅ Cleans up on process exit

Binary Search Order:

1. CAPISCIO_BINARY environment variable (if set)
2. capiscio-core/bin/capiscio relative to SDK (development mode)
3. System PATH (capiscio-core command)
4. Previously downloaded binary in ~/.capiscio/bin/
5. Auto-download from GitHub releases (latest compatible version)

How It Works

1. The Handshake

CapiscIO enforces the A2A Trust Protocol:

• Sender: Signs the request body (JWS + Body Hash).
• Receiver: Verifies the signature and re-hashes the body to ensure integrity.

2. The "Customs Officer"

The SimpleGuard acts as a local authority. It manages your agent's "Passport" (Agent Card) and verifies the "Visas" (Tokens) of incoming requests.

3. Go Core Integration

The SDK delegates validation to capiscio-core (Go) via gRPC for:

• Consistent behavior across all CapiscIO SDKs
• High performance validation (Go's speed + Python's flexibility)
• Single source of truth for validation rules

4. Telemetry

Every response includes a Server-Timing header showing exactly how fast the verification was:

Server-Timing: capiscio-auth;dur=0.618;desc="CapiscIO Verification"

Documentation

• Official Documentation
• A2A Protocol Spec

License

Apache License 2.0 - see LICENSE for details.