Capability Decorators

AIM provides the @agent.perform_action decorator for automatic capability verification. Risk levels are automatically detected from capability patterns, or you can override manually. Add jit_access=True for approval-gated access to sensitive operations.

@agent.perform_action Options

Feature	`@agent.perform_action(...)`	`@agent.perform_action(..., jit_access=True)`
Execution	Immediate after verification	Waits for approval
Blocking	No	Yes (until approved)
Key Parameter	`capability` + `risk_level`	`capability` + `jit_access=True`
Use Case	Standard operations, monitoring	Sensitive/destructive ops
Analogy	Badge scan (automated)	JIT access / break-glass

@agent.perform_action() — Standard Verification

Use @agent.perform_action for capability-based verification. Functions execute immediately after verification and are logged with their risk level. Perfect for audit trails with automatic verification.

Parameters

Parameter	Type	Description
capability	str (required)	Capability type in namespace:action format (e.g., "db:read")
risk_level	str (optional)	Auto-detected from capability pattern. Override with "low", "medium", "high", or "critical"
resource	str (optional)	Resource being accessed (e.g., "database:users")
auto_register	bool (default: True)	NEW: Automatically register capability on first use. Set to `False` to require pre-registration.
jit_access	bool (default: False)	Enable Just-In-Time access requiring admin approval before execution.
timeout_seconds	int (default: 300)	Max time to wait for JIT approval (only applies when jit_access=True).

🔄 Auto-Registration (New Feature)

By default, capabilities are automatically registered on first use. This means you can use the decorator without pre-registering capabilities — they will be created when first invoked.

auto_register=True (Default)

✓ Capability registered automatically on first call
✓ Great for development and rapid iteration
✓ Works with MONITORING enforcement mode

auto_register=False

✓ Requires capability to be pre-registered
✓ Stricter control for production
✓ Use with STRICT enforcement mode

🎯 Risk Level Auto-Detection

You no longer need to manually specify risk_level! AIM automatically detects the appropriate risk level based on capability patterns:

Detection Priority

Explicit override (if provided)
Specific capability mapping (e.g., "user:delete")
Action patterns (e.g., ":delete" → high)
Namespace patterns (e.g., "payment:" → critical)
Default → medium

Example Auto-Detections

weather:fetch → low

db:write → medium

file:delete → high

payment:process → critical

💡 Tip: Use the explain_risk_detection() function to understand why a capability was assigned a specific risk level.

🟢 Low

• Basic logging
• No trust impact
• Never triggers alerts

🟡 Medium

• Pattern monitoring
• Minor trust impact on anomaly
• Alerts on unusual patterns

🟠 High

• Detailed audit logging
• Can decrease trust score
• May trigger alerts

🔴 Critical

• Full audit trail
• Requires high trust score
• Always reviewed

from aim_sdk import secure

agent = secure("my-agent")

# Risk levels are AUTO-DETECTED from capability patterns!
@agent.perform_action(capability="weather:fetch")  # Auto: low (read-only namespace)
def fetch_weather(city: str) -> dict:
    """Low-risk: logged, doesn't affect trust score"""
    return weather_api.get(city)

@agent.perform_action(capability="notification:send", resource="notifications")  # Auto: high
def send_notification(user_id: str, message: str):
    """High-risk: notification namespace + send action"""
    return notifications.send(user_id, message)

@agent.perform_action(capability="db:delete", resource="users")  # Auto: high
def delete_user_data(user_id: str):
    """High-risk: delete action is inherently high-risk"""
    return database.delete_user(user_id)

# Override when you know better than the auto-detection
@agent.perform_action(capability="api:internal", risk_level="critical")
def internal_admin_call():
    """Override: internal API that needs critical monitoring"""
    return admin_api.call()

# Call the functions normally - verification is automatic
weather = fetch_weather("San Francisco")
send_notification("user-123", "Hello!")

@agent.perform_action(..., jit_access=True) — JIT Access with Approval

Add jit_access=True for sensitive operations that require admin approval before execution. The function blocks until AIM grants permission — like requesting just-in-time (JIT) access to a sensitive resource.

Parameters

Parameter	Type	Description
capability	str	Capability type in namespace:action format (e.g., "payment:refund")
resource	str (optional)	Resource being accessed
context	dict (optional)	Additional context for the action
timeout_seconds	int	Max time to wait for approval (default: 300)

Execution Flow

1. Function called → Request sent to AIM

2. AIM evaluates → Trust score, capabilities, policies

3. Auto-approved OR Needs admin approval

4. If pending → Waits (up to timeout_seconds)

5. Approved → Function executes

6. Denied → ActionDeniedError raised

7. Result logged back to AIM

from aim_sdk import secure
from aim_sdk.exceptions import CapabilityDeniedError

agent = secure("my-agent")

@agent.perform_action(capability="payment:refund", resource="stripe", jit_access=True)
def process_refund(order_id: str, amount: float):
    """
    JIT Access: Waits for AIM approval before executing.
    May require admin approval based on trust score.
    """
    return stripe.refund(order_id, amount)

@agent.perform_action(capability="db:delete", resource="users_table", jit_access=True, timeout_seconds=300)
def purge_inactive_users():
    """
    Destructive operation: Always requires approval.
    Will wait up to 5 minutes for admin to approve.
    """
    return db.purge_inactive()

# When called, these functions will:
# 1. Request verification from AIM
# 2. Wait for approval (if required)
# 3. Execute only if approved
# 4. Log the result back to AIM
try:
    result = process_refund("order-123", 99.99)
except CapabilityDeniedError as e:
    print(f"Refund blocked: {e.reason}")

When to Use Which?

Use @agent.perform_action (standard) for:

✓ Read operations (fetching data, queries)
✓ Non-destructive writes (caching, logging)
✓ Operations that shouldn't block
✓ High-frequency capabilities
✓ When you need audit trails without gates

Use @agent.perform_action (jit_access=True) for:

✓ Financial transactions (payments, refunds)
✓ Destructive operations (delete, purge)
✓ Permission changes (granting access)
✓ External system modifications
✓ When human approval may be required

Combined Example

from aim_sdk import secure

agent = secure("operations-agent")

# Standard capability verification (immediate execution after verification)
@agent.perform_action(capability="config:read", risk_level="low")
def read_config():
    """Executes immediately after verification, logged for audit"""
    return load_config()

@agent.perform_action(capability="cache:update", risk_level="medium", resource="cache")
def update_cache(key: str, value: str):
    """Executes immediately after verification, monitored for patterns"""
    return cache.set(key, value)

# JIT Access: Sensitive ops requiring admin approval
@agent.perform_action(capability="admin:modify_permissions", resource="rbac", jit_access=True)
def grant_admin_access(user_id: str):
    """Waits for approval - sensitive operation"""
    return rbac.grant_admin(user_id)

@agent.perform_action(capability="billing:charge", resource="stripe", jit_access=True, timeout_seconds=60)
def charge_customer(customer_id: str, amount: float):
    """Waits for approval - financial operation"""
    return stripe.charge(customer_id, amount)

Async Support

Both decorators work seamlessly with async functions:

from aim_sdk import secure
import asyncio

agent = secure("async-agent")

# @agent.perform_action works with async functions
@agent.perform_action(capability="data:fetch", risk_level="low")
async def fetch_data(url: str):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

@agent.perform_action(capability="external:api_call", resource="partner-api", jit_access=True)
async def call_partner_api(endpoint: str, payload: dict):
    async with aiohttp.ClientSession() as session:
        async with session.post(endpoint, json=payload) as response:
            return await response.json()

# Usage
async def main():
    data = await fetch_data("https://api.example.com/data")
    result = await call_partner_api("/process", {"data": data})

Error Handling

from aim_sdk import secure
from aim_sdk.exceptions import CapabilityDeniedError, VerificationError, TimeoutError

agent = secure("error-handling-demo")

@agent.perform_action(capability="db:write", resource="production", jit_access=True)
def write_to_production(data: dict):
    return db.write(data)

# Proper error handling
try:
    write_to_production({"key": "value"})
except CapabilityDeniedError as e:
    # Capability was explicitly denied by AIM
    print(f"Capability denied: {e.reason}")
    print(f"Required capability: {e.required_capability}")
    print(f"Current trust score: {e.trust_score}")
except TimeoutError as e:
    # Approval took too long
    print(f"Approval timeout: No response within {e.timeout}s")
except VerificationError as e:
    # Verification system error
    print(f"Verification failed: {e}")

Dashboard Visibility

All decorator-tracked actions appear in the AIM Dashboard under your agent's Activity tab:

• Green badge — Low risk actions
• Yellow badge — Medium risk actions
• Orange badge — High risk actions
• Red badge — Critical / approval-required actions

Navigate to: Dashboard → Agents → [Your Agent] → Activity

Next Steps

Python SDK Reference →CBAC Documentation →Trust Scoring →Activity Monitoring →