pip install capiscio-sdkfrom capiscio_sdk import secure
agent = secure(MyAgentExecutor())from capiscio_sdk import SecurityConfig
# Development (permissive)
SecurityConfig.development()
# Production (balanced) - default
SecurityConfig.production()
# Strict (maximum security)
SecurityConfig.strict()
# From environment variables
SecurityConfig.from_env()📖 See Configuration Guide for all options.
agent = secure(MyAgentExecutor())config = SecurityConfig.production()
agent = CapiscIOSecurityExecutor(MyAgentExecutor(), config)@secure_agent(config=SecurityConfig.production())
class MyAgent(AgentExecutor):
pass| Validator | Purpose | Performance |
|---|---|---|
| MessageValidator | Schema & structure | ~1-5ms |
| ProtocolValidator | A2A compliance | ~1-5ms |
| SignatureValidator | JWT/JWS verification | ~10-50ms (cached) |
| SemverValidator | Version compatibility | <1ms |
| URLSecurityValidator | SSRF prevention | ~1-5ms |
| AgentCardValidator | Discovery metadata | ~10-50ms (cached) |
| CertificateValidator | TLS/SSL validation | ~50-200ms (cached) |
config.downstream.validate_schema = True
config.downstream.verify_signatures = True
config.downstream.require_signatures = False
config.downstream.check_protocol_compliance = True
config.downstream.enable_rate_limiting = True
config.downstream.rate_limit_requests_per_minute = 60config.upstream.validate_agent_cards = True
config.upstream.verify_signatures = True
config.upstream.require_signatures = False
config.upstream.test_endpoints = False
config.upstream.cache_validation = True
config.upstream.cache_timeout = 3600config.fail_mode = "block" # "block" | "monitor" | "log"
config.log_validation_failures = True
config.timeout_ms = 5000from capiscio_sdk.errors import (
CapiscIOValidationError,
CapiscIOSignatureError,
CapiscIORateLimitError,
CapiscIOUpstreamError,
)
try:
await agent.execute(context, event_queue)
except CapiscIOValidationError as e:
print(f"Validation failed: {e.message}")
print(f"Errors: {e.errors}")
except CapiscIORateLimitError as e:
print(f"Rate limit exceeded, retry after {e.retry_after_seconds}s")# Downstream
export CAPISCIO_VALIDATE_SCHEMA=true
export CAPISCIO_VERIFY_SIGNATURES=true
export CAPISCIO_REQUIRE_SIGNATURES=false
export CAPISCIO_RATE_LIMITING=true
export CAPISCIO_RATE_LIMIT_RPM=60
# Upstream
export CAPISCIO_VALIDATE_UPSTREAM=true
export CAPISCIO_CACHE_VALIDATION=true
# General
export CAPISCIO_FAIL_MODE=block
export CAPISCIO_TIMEOUT_MS=5000import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("capiscio_sdk")agent = secure(MyAgentExecutor(), SecurityConfig.development())config = SecurityConfig.production()
config.downstream.rate_limit_requests_per_minute = 120
agent = secure(MyAgentExecutor(), config)config = SecurityConfig.strict()
config.upstream.test_endpoints = True
agent = secure(MyAgentExecutor(), config)config = SecurityConfig.production()
config.fail_mode = "monitor"
agent = secure(MyAgentExecutor(), config)