Python SDK API Reference

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

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()

Register this agent with the ASCEND platform.

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']}")

---

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")

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)

---

Constants

# SDK Version
AscendClient.VERSION = "2.0.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

---

See Also

• Quickstart Guide
• MCP Integration
• Code Examples