Python SDK API Reference

Complete reference for all classes, methods, and types in the ASCEND Python SDK v2.1.

Installation

pip install ascend-ai-sdk

AscendClient

The main client class for interacting with the ASCEND Platform.

Constructor

from ascend import AscendClient, FailMode

client = AscendClient(
    api_key: str = None,
    agent_id: str = None,
    agent_name: str = None,
    api_url: str = None,
    environment: str = "production",
    fail_mode: str = "closed",
    timeout: int = 5,
    max_retries: int = 3,
    enable_circuit_breaker: bool = True,
    circuit_breaker_threshold: int = 5,
    circuit_breaker_timeout: int = 30,
    signing_secret: str = None,
    debug: bool = False
)

Parameters

Parameter	Type	Default	Description
`api_key`	`str`	`None`	Organization API key. Falls back to `ASCEND_API_KEY` env var.
`agent_id`	`str`	`None`	Unique identifier for this agent. Falls back to `ASCEND_AGENT_ID`.
`agent_name`	`str`	`None`	Human-readable agent name. Falls back to `ASCEND_AGENT_NAME`.
`api_url`	`str`	`https://pilot.owkai.app`	API endpoint URL. Falls back to `ASCEND_API_URL`.
`environment`	`str`	`"production"`	Deployment environment (production, staging, development).
`fail_mode`	`str`	`"closed"`	Behavior when ASCEND is unreachable: "closed" (block) or "open" (allow).
`timeout`	`int`	`5`	Request timeout in seconds.
`max_retries`	`int`	`3`	Maximum retry attempts for failed requests.
`enable_circuit_breaker`	`bool`	`True`	Enable circuit breaker pattern.
`circuit_breaker_threshold`	`int`	`5`	Failures before opening circuit.
`circuit_breaker_timeout`	`int`	`30`	Seconds before attempting recovery.
`signing_secret`	`str`	`None`	HMAC secret for request signing.
`debug`	`bool`	`False`	Enable debug logging.

Methods

register()

def register(
    agent_type: str = "supervised",
    capabilities: List[str] = None,
    allowed_resources: List[str] = None,
    metadata: Dict = None
) -> Dict[str, Any]

Parameters:

Parameter	Type	Default	Description
`agent_type`	`str`	`"supervised"`	Agent type: "supervised", "autonomous", "mcp_server"
`capabilities`	`List[str]`	`None`	List of capabilities (e.g., ["database.query", "file.write"])
`allowed_resources`	`List[str]`	`None`	Resources this agent can access
`metadata`	`Dict`	`None`	Additional agent metadata

Returns: Dict[str, Any] - Registration confirmation with agent details

Example:

result = client.register(
    agent_type="automation",
    capabilities=["database.query", "file.read", "api.call"],
    allowed_resources=["production_db", "customer_api"],
    metadata={"version": "1.0.0", "owner": "platform-team"}
)

print(f"Registered: {result['agent_id']}")
print(f"Trust level: {result.get('trust_level')}")

evaluate_action()

Evaluate an action against ASCEND policies.

def evaluate_action(
    action_type: str,
    resource: str,
    parameters: Dict[str, Any] = None,
    context: Dict[str, Any] = None,
    resource_id: str = None,
    risk_indicators: Dict[str, Any] = None,
    wait_for_decision: bool = True,
    timeout: int = None
) -> AuthorizationDecision

Parameters:

Parameter	Type	Default	Description
`action_type`	`str`	Required	Action type in "category.action" format (e.g., "database.query")
`resource`	`str`	Required	Resource identifier (e.g., "production_db")
`parameters`	`Dict`	`None`	Action-specific parameters
`context`	`Dict`	`None`	Additional context for risk scoring
`resource_id`	`str`	`None`	Specific resource instance ID
`risk_indicators`	`Dict`	`None`	Risk assessment hints
`wait_for_decision`	`bool`	`True`	Whether to wait for approval if pending
`timeout`	`int`	`None`	Timeout for waiting (seconds)

Returns: AuthorizationDecision - Authorization decision with status and metadata

Example:

from ascend import Decision

decision = client.evaluate_action(
    action_type="financial.refund",
    resource="stripe_api",
    parameters={
        "amount": 100.00,
        "currency": "USD",
        "customer_id": "cust_123"
    },
    context={
        "session_id": "sess_abc",
        "user_request": "Process refund for order #456"
    },
    risk_indicators={
        "financial_data": True,
        "amount_threshold": "normal"
    }
)

if decision.decision == Decision.ALLOWED:
    print(f"Action approved: {decision.action_id}")
elif decision.decision == Decision.PENDING:
    print(f"Awaiting approval: {decision.approval_request_id}")
else:
    print(f"Denied: {decision.reason}")

get_action_status()

Get the current status of an action.

def get_action_status(action_id: str) -> AuthorizationDecision

Parameters:

Parameter	Type	Description
`action_id`	`str`	The action ID returned from evaluate_action

Returns: AuthorizationDecision - Current authorization decision status

Example:

status = client.get_action_status("act_abc123")
print(f"Status: {status.decision}")
print(f"Risk score: {status.risk_score}")

wait_for_decision()

Wait for an authorization decision.

def wait_for_decision(
    action_id: str,
    timeout: int = 60,
    poll_interval: float = 2.0
) -> AuthorizationDecision

Parameters:

Parameter	Type	Default	Description
`action_id`	`str`	Required	The action ID to wait for
`timeout`	`int`	`60`	Maximum time to wait in seconds
`poll_interval`	`float`	`2.0`	Time between status checks in seconds

Returns: AuthorizationDecision - Final authorization decision

Raises: TimeoutError - If decision not received within timeout

Example:

# Submit action without waiting
decision = client.evaluate_action(
    action_type="database.delete",
    resource="production_db",
    parameters={"table": "old_logs"},
    wait_for_decision=False
)

if decision.decision == Decision.PENDING:
    # Wait separately with custom timeout
    final_decision = client.wait_for_decision(
        action_id=decision.action_id,
        timeout=300,  # 5 minutes
        poll_interval=5.0
    )

log_action_completed()

Log that an action was completed successfully.

def log_action_completed(
    action_id: str,
    result: Dict[str, Any] = None,
    duration_ms: int = None
) -> Dict[str, Any]

Parameters:

Parameter	Type	Default	Description
`action_id`	`str`	Required	The action ID from evaluate_action
`result`	`Dict`	`None`	Optional result data to log
`duration_ms`	`int`	`None`	How long the action took in milliseconds

Returns: Dict[str, Any] - Confirmation of logging

Example:

import time

decision = client.evaluate_action(
    action_type="database.query",
    resource="production_db",
    parameters={"query": "SELECT COUNT(*) FROM users"}
)

if decision.decision == Decision.ALLOWED:
    start = time.time()
    result = execute_query()
    duration = int((time.time() - start) * 1000)

    client.log_action_completed(
        action_id=decision.action_id,
        result={"row_count": result["count"]},
        duration_ms=duration
    )

log_action_failed()

Log that an action failed.

def log_action_failed(
    action_id: str,
    error: Dict[str, Any],
    duration_ms: int = None
) -> Dict[str, Any]

Parameters:

Parameter	Type	Default	Description
`action_id`	`str`	Required	The action ID from evaluate_action
`error`	`Dict`	Required	Error details (code, message, etc.)
`duration_ms`	`int`	`None`	How long before failure in milliseconds

Returns: Dict[str, Any] - Confirmation of logging

Example:

try:
    result = execute_query()
except Exception as e:
    client.log_action_failed(
        action_id=decision.action_id,
        error={
            "code": "EXECUTION_ERROR",
            "message": str(e),
            "type": type(e).__name__
        }
    )
    raise

check_approval()

Check the status of an approval request.

def check_approval(approval_request_id: str) -> Dict[str, Any]

Parameters:

Parameter	Type	Description
`approval_request_id`	`str`	The ID returned when action is pending_approval

Returns: Dict[str, Any] with keys:

approved: Boolean - Whether approved
denied: Boolean - Whether denied
pending: Boolean - Whether still pending
approver: String - Who approved/denied
timestamp: String - When decision was made
comments: String - Any comments from approver

Example:

if decision.decision == Decision.PENDING:
    while True:
        status = client.check_approval(decision.approval_request_id)

        if status["approved"]:
            execute_action()
            break
        elif status["denied"]:
            print(f"Denied by {status['approver']}: {status['comments']}")
            break

        time.sleep(5)

configure_webhook()

Configure webhook for real-time notifications.

def configure_webhook(
    url: str,
    events: List[str],
    secret: str = None
) -> Dict[str, Any]

Parameters:

Parameter	Type	Default	Description
`url`	`str`	Required	HTTPS endpoint to receive webhooks
`events`	`List[str]`	Required	List of events to subscribe to
`secret`	`str`	`None`	Shared secret for signature verification

Valid Events:

action.approved
action.denied
action.pending
policy.violation
agent.trust_changed

Returns: Dict[str, Any] - Webhook configuration details

Example:

config = client.configure_webhook(
    url="https://your-app.com/webhooks/ascend",
    events=["action.approved", "action.denied", "policy.violation"],
    secret="whsec_your_secret_here"
)

print(f"Webhook ID: {config['webhook_id']}")

test_connection()

Test API connectivity and authentication.

def test_connection() -> Dict[str, Any]

Returns: Dict[str, Any] with keys:

status: "connected" or "error"
api_version: API version string
environment: Detected environment
agent_id: Configured agent ID
fail_mode: Current fail mode

Example:

status = client.test_connection()

if status["status"] == "connected":
    print(f"Connected to {status['api_version']}")
else:
    print(f"Connection error: {status['error']}")

Enterprise Methods (v2.1)

is_blocked()

Check if the kill switch is active for this agent.

def is_blocked() -> KillSwitchStatus

Returns: KillSwitchStatus - Kill switch status with active, reason, and blocked_at fields

Example:

status = client.is_blocked()
if status.active:
    print(f"Agent blocked: {status.reason}")
    # Halt all operations

start_kill_switch_polling()

Start background polling for kill switch changes.

def start_kill_switch_polling(interval: int = 30) -> None

Parameters:

Parameter	Type	Default	Description
`interval`	`int`	`30`	Polling interval in seconds

Example:

# Start polling every 60 seconds
client.start_kill_switch_polling(interval=60)

# ... agent runs ...

# Stop when shutting down
client.stop_kill_switch_polling()

evaluate_actions()

Evaluate multiple actions concurrently (bulk evaluation).

def evaluate_actions(
    actions: List[AgentAction],
    max_concurrent: int = 5
) -> BulkEvaluationResult

Parameters:

Parameter	Type	Default	Description
`actions`	`List[AgentAction]`	Required	Actions to evaluate (max 50)
`max_concurrent`	`int`	`5`	Max concurrent evaluations

Returns: BulkEvaluationResult - Results with succeeded, failed, total, and per-action results

Example:

from ascend import AgentAction

actions = [
    AgentAction(agent_id="bot-1", agent_name="Bot", action_type="data.read", resource="db1"),
    AgentAction(agent_id="bot-1", agent_name="Bot", action_type="data.write", resource="db2"),
]

result = client.evaluate_actions(actions)
print(f"{result.succeeded}/{result.total} succeeded")

for r in result.results:
    if r.succeeded:
        print(f"  {r.action_type} on {r.resource}: {r.decision.decision}")
    else:
        print(f"  {r.action_type} on {r.resource}: FAILED - {r.error}")

evaluate_policy()

Evaluate an action against a specific policy.

def evaluate_policy(
    policy_id: str,
    action_type: str,
    resource: str,
    parameters: Dict[str, Any] = None
) -> PolicyEvaluationResult

Returns: PolicyEvaluationResult - Policy match result with allowed, policy_id, matched_rules, risk_score

Example:

result = client.evaluate_policy(
    policy_id="pol_financial_controls",
    action_type="financial.transfer",
    resource="stripe_api",
    parameters={"amount": 5000}
)

if not result.allowed:
    print(f"Blocked by rules: {result.matched_rules}")

get_resource_classification()

Get security classification for a resource.

def get_resource_classification(resource_type: str) -> ResourceClassification

Returns: ResourceClassification - Classification with resource_type, sensitivity, compliance_tags, access_level

Example:

classification = client.get_resource_classification("database")
print(f"Sensitivity: {classification.sensitivity}")
print(f"Compliance: {classification.compliance_tags}")

query_audit_log()

Query the audit log for this agent.

def query_audit_log(
    start_date: str = None,
    end_date: str = None,
    action_type: str = None,
    limit: int = 100
) -> AuditLogResponse

Returns: AuditLogResponse - Audit log with events list and total_count

Example:

audit = client.query_audit_log(
    start_date="2026-03-01",
    end_date="2026-03-28",
    action_type="financial.transfer",
    limit=50
)

for event in audit.events:
    print(f"{event.timestamp}: {event.action_type} - {event.decision}")

heartbeat()

Send a heartbeat signal to ASCEND (fire-and-forget).

def heartbeat() -> None

Example:

# Send periodic heartbeats to signal agent health
import threading

def heartbeat_loop():
    while running:
        client.heartbeat()
        time.sleep(30)

threading.Thread(target=heartbeat_loop, daemon=True).start()

deregister()

Deregister this agent from ASCEND.

def deregister() -> None

Example:

# Clean shutdown
client.stop_kill_switch_polling()
client.deregister()

get_agent_status()

Get the health status of this agent.

def get_agent_status() -> AgentHealthStatus

Returns: AgentHealthStatus - Status with agent_id, status, last_heartbeat, uptime_seconds

Example:

status = client.get_agent_status()
print(f"Status: {status.status}")
print(f"Uptime: {status.uptime_seconds}s")

list_webhooks()

List all configured webhooks.

def list_webhooks() -> List[Webhook]

Returns: List[Webhook] - List of webhook configurations

test_webhook()

Test a webhook endpoint.

def test_webhook(webhook_id: str) -> WebhookTestResult

Returns: WebhookTestResult - Test result with success, status_code, response_time_ms

Example:

webhooks = client.list_webhooks()
for wh in webhooks:
    result = client.test_webhook(wh.webhook_id)
    print(f"Webhook {wh.url}: {'OK' if result.success else 'FAIL'} ({result.response_time_ms}ms)")

AuthorizationDecision

Response object from action evaluation.

Properties

Property	Type	Description
`action_id`	`str`	Unique action identifier
`decision`	`Decision`	Decision status (ALLOWED, DENIED, PENDING)
`risk_score`	`int`	Risk score 0-100
`risk_level`	`RiskLevel`	Risk level (LOW, MEDIUM, HIGH, CRITICAL)
`reason`	`str`	Explanation of decision
`policy_violations`	`List[str]`	List of violated policies
`conditions`	`List[str]`	Conditions applied to approval
`approval_request_id`	`str`	ID for tracking approval (if pending)
`required_approvers`	`List[str]`	Who needs to approve
`expires_at`	`str`	When approval expires
`approved_by`	`str`	Who approved (if approved)
`comments`	`str`	Approval/denial comments
`execution_allowed`	`bool`	Convenience boolean for allowed state
`metadata`	`Dict`	Additional metadata

Decision Enum

Authorization decision values.

from ascend import Decision

Decision.ALLOWED   # Action is allowed to proceed
Decision.DENIED    # Action is denied
Decision.PENDING   # Action is pending human approval

FailMode Enum

Fail mode configuration.

from ascend import FailMode

FailMode.CLOSED  # Block actions when ASCEND is unreachable (recommended)
FailMode.OPEN    # Allow actions when ASCEND is unreachable

RiskLevel Enum

Risk assessment levels.

from ascend import RiskLevel

RiskLevel.LOW       # 0-44 risk score
RiskLevel.MEDIUM    # 45-69 risk score
RiskLevel.HIGH      # 70-84 risk score
RiskLevel.CRITICAL  # 85-100 risk score

Exceptions

OWKAIError

Base exception for all SDK errors.

class OWKAIError(Exception):
    message: str       # Error message
    error_code: str    # Error code for programmatic handling
    details: Dict      # Additional error details

AuthenticationError

Raised when authentication fails (invalid/expired API key).

from ascend import AuthenticationError

try:
    client.evaluate_action(...)
except AuthenticationError as e:
    print(f"Auth failed: {e.message}")

AuthorizationError

Raised when an action is denied by policy.

from ascend import AuthorizationError

try:
    client.evaluate_action(...)
except AuthorizationError as e:
    print(f"Denied: {e.message}")
    print(f"Violations: {e.policy_violations}")
    print(f"Risk score: {e.risk_score}")

TimeoutError

Raised when an operation times out.

from ascend import TimeoutError

try:
    client.wait_for_decision(action_id, timeout=60)
except TimeoutError as e:
    print(f"Timed out after {e.timeout_seconds}s")

RateLimitError

Raised when rate limit is exceeded.

from ascend import RateLimitError

try:
    client.evaluate_action(...)
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
    time.sleep(e.retry_after)

ConnectionError

Raised when connection to API fails.

from ascend import ConnectionError

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

CircuitBreakerOpen

Raised when circuit breaker is open.

from ascend import CircuitBreakerOpen

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

ConfigurationError

Raised when client configuration is invalid (e.g., non-HTTPS URL for remote hosts).

from ascend import ConfigurationError

try:
    client = AscendClient(api_key="key", api_url="http://remote-host.com")
except ConfigurationError as e:
    print(f"Config error: {e.message}")  # TLS required

KillSwitchError

Raised when the kill switch is active and an action is attempted.

from ascend import KillSwitchError

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

ValidationError

Raised when input validation fails.

from ascend import ValidationError

try:
    client.configure_webhook(url="http://...", events=[])
except ValidationError as e:
    print(f"Validation failed: {e.message}")
    print(f"Field errors: {e.field_errors}")

Type Reference

AgentAction

@dataclass
class AgentAction:
    agent_id: str
    agent_name: str
    action_type: str
    resource: str
    resource_id: Optional[str] = None
    action_details: Optional[Dict[str, Any]] = None
    context: Optional[Dict[str, Any]] = None
    risk_indicators: Optional[Dict[str, Any]] = None

RiskIndicators

@dataclass
class RiskIndicators:
    pii_involved: bool = False
    financial_data: bool = False
    external_transfer: bool = False
    data_sensitivity: str = "low"
    amount_threshold: Optional[str] = None
    compliance_flags: List[str] = field(default_factory=list)

ActionContext

@dataclass
class ActionContext:
    user_request: Optional[str] = None
    session_id: Optional[str] = None
    ip_address: Optional[str] = None
    user_agent: Optional[str] = None
    timestamp: Optional[str] = None
    custom_fields: Dict[str, Any] = field(default_factory=dict)

KillSwitchStatus

@dataclass
class KillSwitchStatus:
    active: bool = False
    reason: Optional[str] = None
    blocked_at: Optional[str] = None
    blocked_by: Optional[str] = None

PolicyEvaluationResult

@dataclass
class PolicyEvaluationResult:
    allowed: bool
    policy_id: str
    matched_rules: List[str]
    risk_score: int = 0
    details: Optional[Dict[str, Any]] = None

ResourceClassification

@dataclass
class ResourceClassification:
    resource_type: str
    sensitivity: str
    compliance_tags: List[str]
    access_level: str

AgentHealthStatus

@dataclass
class AgentHealthStatus:
    agent_id: str
    status: str
    last_heartbeat: Optional[str] = None
    uptime_seconds: Optional[int] = None

BulkEvaluationResult

@dataclass
class BulkEvaluationResult:
    results: List[BulkActionResult]
    succeeded: int
    failed: int
    total: int

Webhook

@dataclass
class Webhook:
    webhook_id: str
    url: str
    events: List[str]
    active: bool = True
    created_at: Optional[str] = None

Constants

# SDK Version
AscendClient.VERSION = "2.1.0"

# Default API URL
AscendClient.DEFAULT_API_URL = "https://pilot.owkai.app"

# Default timeout
AscendClient.DEFAULT_TIMEOUT = 5

# Default max retries
AscendClient.DEFAULT_MAX_RETRIES = 3

Installation​

AscendClient​

Constructor​

Parameters​

Methods​

register()​

evaluate_action()​

get_action_status()​

wait_for_decision()​

log_action_completed()​

log_action_failed()​

check_approval()​

configure_webhook()​

test_connection()​

Enterprise Methods (v2.1)​

is_blocked()​

start_kill_switch_polling()​

evaluate_actions()​

evaluate_policy()​

get_resource_classification()​

query_audit_log()​

heartbeat()​

deregister()​

get_agent_status()​

list_webhooks()​

test_webhook()​

AuthorizationDecision​

Properties​

Decision Enum​

FailMode Enum​

RiskLevel Enum​

Exceptions​

OWKAIError​

AuthenticationError​

AuthorizationError​

TimeoutError​

RateLimitError​

ConnectionError​

CircuitBreakerOpen​

ConfigurationError​

KillSwitchError​

ValidationError​

Type Reference​

AgentAction​

RiskIndicators​

ActionContext​

KillSwitchStatus​

PolicyEvaluationResult​

ResourceClassification​

AgentHealthStatus​

BulkEvaluationResult​

Webhook​

Constants​

See Also​

Installation

AscendClient

Constructor

Parameters

Methods

register()

evaluate_action()

get_action_status()

wait_for_decision()

log_action_completed()

log_action_failed()

check_approval()

configure_webhook()

test_connection()

Enterprise Methods (v2.1)

is_blocked()

start_kill_switch_polling()

evaluate_actions()

evaluate_policy()

get_resource_classification()

query_audit_log()

heartbeat()

deregister()

get_agent_status()

list_webhooks()

test_webhook()

AuthorizationDecision

Properties

Decision Enum

FailMode Enum

RiskLevel Enum

Exceptions

OWKAIError

AuthenticationError

AuthorizationError

TimeoutError

RateLimitError

ConnectionError

CircuitBreakerOpen

ConfigurationError

KillSwitchError

ValidationError

Type Reference

AgentAction

RiskIndicators

ActionContext

KillSwitchStatus

PolicyEvaluationResult

ResourceClassification

AgentHealthStatus

BulkEvaluationResult

Webhook

Constants

See Also