Error Handling

Handle errors effectively when using the ASCEND Python SDK.

Exception Hierarchy

The SDK provides typed exceptions for every error scenario:

OWKAIError (base)
├── AuthenticationError      # 401 — Invalid/expired API key
├── AuthorizationError       # 403 — Insufficient permissions
├── ValidationError          # 400/422 — Invalid input
├── RateLimitError           # 429 — Rate limit exceeded
├── TimeoutError             # Request timeout
├── ConnectionError          # Network connection failure
├── CircuitBreakerOpen       # Circuit breaker is open
├── ConfigurationError       # Invalid client configuration
├── KillSwitchError          # Kill switch is active
├── NotFoundError            # 404 — Resource not found
├── ConflictError            # 409 — Conflict
└── ServerError              # 5xx — Server error

Basic Error Handling

from ascend import (
    AscendClient,
    AuthenticationError,
    AuthorizationError,
    ValidationError,
    RateLimitError,
    TimeoutError,
    ConnectionError,
    CircuitBreakerOpen,
    ConfigurationError,
    KillSwitchError,
)

client = AscendClient(
    api_key="owkai_live_xxxxxxxxxxxx",
    agent_id="my-agent",
    agent_name="My Agent",
)

try:
    decision = client.evaluate_action(
        action_type="database.query",
        resource="production_db",
    )

except AuthenticationError as e:
    print(f"Invalid API key: {e.message}")

except AuthorizationError as e:
    print(f"Denied: {e.message}")
    print(f"Violations: {e.policy_violations}")

except ValidationError as e:
    print(f"Invalid input: {e.message}")
    print(f"Field errors: {e.field_errors}")

except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")

except TimeoutError as e:
    print(f"Timed out after {e.timeout_seconds}s")

except ConnectionError as e:
    print(f"Connection failed: {e.message}")

except CircuitBreakerOpen as e:
    print(f"Service unavailable. Recovery in {e.recovery_time}s")

except KillSwitchError as e:
    print(f"Agent blocked: {e.message}")

except ConfigurationError as e:
    print(f"Config error: {e.message}")

Configuration Errors

Missing API Key

try:
    client = AscendClient(agent_id="test", agent_name="Test")
except AuthenticationError:
    print("Set ASCEND_API_KEY or pass api_key parameter")

Invalid URL (TLS Enforcement)

try:
    client = AscendClient(
        api_key="owkai_live_xxxxxxxxxxxx",
        api_url="http://remote-host.com",  # HTTP not allowed
        agent_id="test",
        agent_name="Test",
    )
except ConfigurationError:
    print("Use https:// for remote hosts")

HTTP is only allowed for localhost and 127.0.0.1.

Input Validation

The SDK validates inputs before making network calls:

try:
    # Empty action_type rejected
    client.evaluate_action(action_type="", resource="db")
except ValidationError as e:
    print(f"Validation failed: {e.message}")

try:
    # None resource rejected
    client.evaluate_action(action_type="data.read", resource=None)
except ValidationError as e:
    print(f"Validation failed: {e.message}")

Rate Limiting

import time

try:
    decision = client.evaluate_action(
        action_type="database.query",
        resource="production_db",
    )
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
    time.sleep(e.retry_after)
    # Retry
    decision = client.evaluate_action(
        action_type="database.query",
        resource="production_db",
    )

Automatic Retries

The SDK automatically retries transient errors (5xx, timeouts, connection errors) with exponential backoff and jitter. Set max_retries=0 to disable.

Circuit Breaker

The circuit breaker opens after repeated failures to prevent cascading issues:

from ascend import AscendClient, CircuitBreakerOpen

client = AscendClient(
    api_key="owkai_live_xxxxxxxxxxxx",
    agent_id="my-agent",
    agent_name="My Agent",
    enable_circuit_breaker=True,
    circuit_breaker_threshold=5,   # Open after 5 failures
    circuit_breaker_timeout=30,    # Recovery attempt after 30s
)

try:
    decision = client.evaluate_action(
        action_type="database.query",
        resource="production_db",
    )
except CircuitBreakerOpen as e:
    print(f"Service unavailable. Recovery in {e.recovery_time}s")
    # Fall back to cached decision or fail-mode behavior

Fail Mode Behavior

When the ASCEND API is unreachable, behavior depends on the configured fail mode:

from ascend import AscendClient, FailMode, Decision

# Fail-closed: raises error when API is unreachable
closed_client = AscendClient(
    api_key="owkai_live_xxxxxxxxxxxx",
    agent_id="agent",
    agent_name="Agent",
    fail_mode=FailMode.CLOSED,
)

# Fail-open: returns synthetic ALLOWED decision
open_client = AscendClient(
    api_key="owkai_live_xxxxxxxxxxxx",
    agent_id="agent",
    agent_name="Agent",
    fail_mode=FailMode.OPEN,
)

# With fail-open, unreachable API returns:
# decision.decision == Decision.ALLOWED
# decision.conditions == ["fail_open_mode"]

Comprehensive Error Handler

from ascend import AscendClient, Decision, OWKAIError

def safe_evaluate(client, action_type, resource, **kwargs):
    """Evaluate action with comprehensive error handling."""
    try:
        decision = client.evaluate_action(
            action_type=action_type,
            resource=resource,
            **kwargs,
        )
        return {"success": True, "decision": decision}

    except AuthenticationError:
        return {"success": False, "error": "auth_failed", "retry": False}

    except ValidationError as e:
        return {"success": False, "error": f"invalid_input: {e.message}", "retry": False}

    except RateLimitError as e:
        return {"success": False, "error": "rate_limited", "retry": True, "retry_after": e.retry_after}

    except (TimeoutError, ConnectionError):
        return {"success": False, "error": "network", "retry": True}

    except CircuitBreakerOpen:
        return {"success": False, "error": "circuit_open", "retry": True}

    except KillSwitchError:
        return {"success": False, "error": "kill_switch", "retry": False}

    except OWKAIError as e:
        return {"success": False, "error": str(e), "retry": False}

Error Properties

All SDK exceptions extend OWKAIError and include:

Property	Type	Description
message	str	Human-readable error message
error_code	str	Machine-readable error code
details	dict	Additional error context

Specific exceptions add extra properties:

Exception	Extra Properties
RateLimitError	retry_after (seconds)
TimeoutError	timeout_seconds
CircuitBreakerOpen	recovery_time (seconds)
AuthorizationError	policy_violations, risk_score
ValidationError	field_errors

Logging Errors

The SDK's structured logger automatically masks sensitive data:

import logging

logging.basicConfig(level=logging.INFO)

# API keys are automatically masked in log output:
# "owkai_live_abc123..." → "owkai_****"
# "Bearer eyJhbGci..." → "Bearer ****"

client = AscendClient(
    api_key="owkai_live_xxxxxxxxxxxx",
    agent_id="my-agent",
    agent_name="My Agent",
    debug=True,  # Enable debug-level logging
)

Best Practices

1. Catch specific exceptions — Handle each error type appropriately rather than catching bare Exception.
2. Use fail-closed in production — FailMode.CLOSED blocks actions when governance is unavailable.
3. Let the SDK retry — The SDK retries transient errors automatically with exponential backoff + jitter.
4. Log action outcomes — Always call log_action_completed() or log_action_failed() after execution.
5. Monitor circuit breaker — Watch for CircuitBreakerOpen errors as a signal of API health issues.

Next Steps

• API Reference — Complete API documentation
• Client Configuration — Configure timeouts and retries
• Agent Actions — Submit and manage actions