Skip to content

CapiscioSecurityExecutor

Full-featured security wrapper that wraps an A2A agent executor with protection.

Overview

CapiscioSecurityExecutor provides:

  • Request validation - Validates incoming requests
  • Signature verification - Verifies JWS signatures on requests
  • Agent card validation - Validates caller's agent card
  • Response signing - Signs all outgoing responses
  • Rate limiting - Optional rate limiting
  • Audit logging - Logs all security events

Basic Usage

from capiscio_sdk import CapiscioSecurityExecutor, SecurityConfig

# Wrap your existing executor
secured_executor = CapiscioSecurityExecutor(
    wrapped_executor=my_agent_executor,
    config=SecurityConfig.production()
)

# Use it like your original executor
result = await secured_executor.execute(context, event_queue)

Wrapper Functions

secure()

Minimal pattern for wrapping an agent:

from capiscio_sdk import secure, SecurityConfig

secured = secure(my_agent, config=SecurityConfig.production())

secure_agent()

Decorator pattern:

from capiscio_sdk import secure_agent, SecurityConfig

@secure_agent(config=SecurityConfig.production())
class MyAgent:
    def execute(self, context, event_queue):
        pass

API Reference

capiscio_sdk.executor.CapiscioSecurityExecutor

Security wrapper for A2A agent executors.

Provides runtime validation, rate limiting, and security checks for A2A agent interactions. Implements the AgentExecutor interface.

__init__ 

__init__(delegate: Any, config: Optional[SecurityConfig] = None)

Initialize security executor.

Parameters:

Name Type Description Default
delegate Any

The agent executor to wrap (must implement AgentExecutor interface)

 required
config Optional[SecurityConfig]

Security configuration (defaults to production preset)

 None

execute async 

execute(context: RequestContext, event_queue: Any) -> None

Execute agent with security checks.

Parameters:

Name Type Description Default
context RequestContext

RequestContext with message and task information

 required
event_queue Any

EventQueue for publishing events

 required

Raises:

Type Description
CapiscioValidationError

If validation fails in block mode
CapiscioRateLimitError

If rate limit exceeded in block mode

cancel async 

cancel(context: RequestContext, event_queue: Any) -> None

Cancel task with passthrough to delegate.

Parameters:

Name Type Description Default
context RequestContext

RequestContext with task to cancel

 required
event_queue Any

EventQueue for publishing cancellation event

 required

validate_agent_card async 

validate_agent_card(url: str) -> ValidationResult

Validate an agent card from a URL.

Parameters:

Name Type Description Default
url str

URL to the agent card or agent root

 required

Returns:

Type Description
ValidationResult

ValidationResult with scores

__getattr__ 

__getattr__(name: str) -> Any

Delegate attribute access to wrapped executor.

Functions

capiscio_sdk.executor.secure 

secure(agent: Any, config: Optional[SecurityConfig] = None) -> CapiscioSecurityExecutor

Wrap an agent executor with security middleware (minimal pattern).

Parameters:

Name Type Description Default
agent Any

Agent executor to wrap

 required
config Optional[SecurityConfig]

Security configuration (defaults to production)

 None

Returns:

Type Description
CapiscioSecurityExecutor

Secured agent executor
Example 
agent = secure(MyAgentExecutor())

capiscio_sdk.executor.secure_agent 

secure_agent(config: Optional[SecurityConfig] = None) -> Callable[[type], Callable[..., CapiscioSecurityExecutor]]

Decorator to secure an agent executor class (decorator pattern).

Parameters:

Name Type Description Default
config Optional[SecurityConfig]

Security configuration (defaults to production)

 None

Returns:

Type Description
Callable[[type], Callable[..., CapiscioSecurityExecutor]]

Decorator function
Example 
@secure_agent(config=SecurityConfig.strict())
class MyAgent:
    def execute(self, message):
        # ... agent logic