Python SDK

Field	Value
Document ID	ASCEND-SDK-011
Version	1.0.0
Last Updated	December 19, 2025
Author	Ascend Engineering Team
Publisher	OW-KAI Technologies Inc.
Classification	Enterprise Client Documentation
Compliance	SOC 2 CC6.1/CC6.2, PCI-DSS 7.1/8.3, HIPAA 164.312, NIST 800-53 AC-2/SI-4

Reading Time: 15 minutes | Skill Level: Intermediate

Overview

The ASCEND Python SDK (ascend-sdk) provides enterprise-grade governance integration for Python-based AI agents. It features automatic retry, circuit breaker patterns, and comprehensive error handling.

Installation

pip install ascend-ai-sdk

Package Info: ascend-ai-sdk on PyPI | Version: 1.0.0 | License: MIT

Requirements

Python 3.8+
requests >= 2.28.0
python-dotenv >= 1.0.0

Verify Installation

import ascend
print(f"ASCEND SDK Version: {ascend.__version__}")

Quick Start

# Source: sdk/ascend-sdk-python/ascend/client.py:64
import os
from ascend import AscendClient, AgentAction

# Initialize client
client = AscendClient(
    api_key=os.environ["ASCEND_API_KEY"],
    base_url="https://pilot.owkai.app",
    timeout=30,
    debug=False
)

# Create an action
action = AgentAction(
    agent_id="financial-advisor-prod",
    agent_name="Financial Advisor Bot",
    action_type="database_read",
    resource="Query customer portfolio",
    tool_name="postgresql",
    resource_id="portfolio_123",
    action_details={"query_type": "balance_check"},
    context={"session_id": "sess_abc123"}
)

# Submit for governance
result = client.submit_action(action)

# Check decision
if result.is_approved():
    print(f"Approved! Proceeding with action...")
    # Execute your action here
elif result.is_denied():
    print(f"Denied: {result.reason}")
elif result.is_pending():
    print(f"Pending approval from: {result.metadata.get('pending_approvers')}")

Client Configuration

Constructor Options

# Source: sdk/ascend-sdk-python/ascend/client.py:92
client = AscendClient(
    api_key="owkai_xxx",              # Required: Your API key
    base_url="https://pilot.owkai.app", # API endpoint
    timeout=30,                        # Request timeout (seconds)
    debug=False                        # Enable debug logging
)

Parameter	Type	Default	Description
`api_key`	str	Required	ASCEND API key (or `ASCEND_API_KEY` env var)
`base_url`	str	`https://pilot.owkai.app`	API endpoint URL
`timeout`	int	30	Request timeout in seconds
`debug`	bool	False	Enable verbose logging

Environment Variables

export ASCEND_API_KEY="owkai_your_key_here"
export ASCEND_API_URL="https://pilot.owkai.app"
export ASCEND_DEBUG="false"

AgentAction Model

Required Fields

# Source: sdk/ascend-sdk-python/ascend/models.py:14
from ascend import AgentAction

action = AgentAction(
    agent_id="my-agent-001",           # Required: Unique agent identifier
    agent_name="My AI Agent",          # Required: Human-readable name
    action_type="database_read",       # Required: Action category
    resource="Description of action",  # Required: What is being done
    tool_name="postgresql"             # Required: Tool/service name
)

All Fields

Field	Type	Required	Description
`agent_id`	str	Yes	Unique agent identifier
`agent_name`	str	Yes	Human-readable agent name
`action_type`	str	Yes	Action category (see below)
`resource`	str	Yes	Description of the action
`tool_name`	str	Yes	Name of tool/service being used
`resource_id`	str	No	Target resource identifier
`action_details`	dict	No	Action-specific parameters
`context`	dict	No	Execution context
`risk_indicators`	dict	No	Pre-computed risk signals

Standard Action Types

Category	Action Types
Database	`database_read`, `database_write`, `database_delete`, `database_schema`
File System	`file_read`, `file_write`, `file_delete`, `file_execute`
Network	`http_request`, `api_call`, `email_send`, `webhook_trigger`
System	`process_spawn`, `config_change`, `credential_access`
AI	`model_inference`, `prompt_injection`, `data_extraction`
Financial	`transaction`, `payment_process`, `refund`

ActionResult Response

Response Fields

# Source: sdk/ascend-sdk-python/ascend/models.py:101
result = client.submit_action(action)

print(f"Action ID: {result.action_id}")
print(f"Status: {result.status}")
print(f"Decision: {result.decision}")
print(f"Risk Score: {result.risk_score}")
print(f"Risk Level: {result.risk_level}")
print(f"Reason: {result.reason}")
print(f"Timestamp: {result.timestamp}")

Field	Type	Description
`action_id`	str	Unique identifier for this action
`status`	str	Current status: `approved`, `denied`, `pending`
`decision`	str	Same as status (for compatibility)
`risk_score`	float	Risk score (0-100)
`risk_level`	str	Risk level: `low`, `medium`, `high`, `critical`
`reason`	str	Explanation of decision
`policy_matched`	str	Name of matched policy
`timestamp`	str	ISO 8601 timestamp
`metadata`	dict	Additional response data

Helper Methods

# Check decision status
if result.is_approved():
    # Action was approved
    execute_action()

elif result.is_denied():
    # Action was blocked
    log_denial(result.reason)

elif result.is_pending():
    # Needs human approval
    wait_or_notify()

Waiting for Decisions

Synchronous Wait

# Source: sdk/ascend-sdk-python/ascend/client.py:474
# Submit action
result = client.submit_action(action)

if result.is_pending():
    # Wait up to 5 minutes for approval
    final_result = client.wait_for_decision(
        action_id=result.action_id,
        timeout_ms=300000,    # 5 minutes
        poll_interval=5.0     # Check every 5 seconds
    )

    if final_result.is_approved():
        print("Approval received!")
        execute_action()

Check Status Later

# Source: sdk/ascend-sdk-python/ascend/client.py:451
# Store action ID
action_id = result.action_id

# ... later in your code ...

# Get current status
current_status = client.get_action_status(action_id)
print(f"Current status: {current_status.status}")

Get Full Action Details

# Source: sdk/ascend-sdk-python/ascend/client.py:430
# Get complete action record with audit trail
full_action = client.get_action(action_id)

print(f"Action: {full_action.action_id}")
print(f"Status: {full_action.status}")
print(f"Metadata: {full_action.metadata}")

Listing Actions

# Source: sdk/ascend-sdk-python/ascend/client.py:529
# List recent actions
result = client.list_actions(
    limit=50,           # Max results per page
    offset=0,           # Pagination offset
    status="pending"    # Filter by status
)

for action in result.actions:
    print(f"{action.action_id}: {action.status}")

# Check for more pages
if result.has_more:
    next_page = client.list_actions(limit=50, offset=50)

Error Handling

Exception Types

# Source: sdk/ascend-sdk-python/ascend/exceptions.py
from ascend.exceptions import (
    AscendError,              # Base exception
    AuthenticationError,      # Invalid/expired API key
    AuthorizationDeniedError, # Action denied by policy
    RateLimitError,           # Too many requests
    TimeoutError,             # Request timeout
    ValidationError,          # Invalid input
    NetworkError,             # Connection failed
    ServerError,              # 5xx server error
    NotFoundError,            # Resource not found
    ConflictError             # Resource conflict
)

Production Error Handling

from ascend import AscendClient, AgentAction
from ascend.exceptions import (
    AuthenticationError,
    NetworkError,
    RateLimitError,
    TimeoutError,
    ValidationError
)

client = AscendClient(api_key=os.environ["ASCEND_API_KEY"])

try:
    result = client.submit_action(action)

    if result.is_approved():
        execute_business_logic()

except AuthenticationError as e:
    logger.error(f"Invalid API key: {e}")
    # Check ASCEND_API_KEY environment variable

except ValidationError as e:
    logger.error(f"Invalid action data: {e}")
    # Check action parameters

except NetworkError as e:
    logger.error(f"Network error: {e}")
    # Implement fallback behavior

except RateLimitError as e:
    logger.warning(f"Rate limited: {e}")
    # Wait and retry, or queue for later

except TimeoutError as e:
    logger.error(f"Request timeout: {e}")
    # Action status is unknown - check later

Context Manager

Use the client as a context manager for automatic cleanup:

# Source: sdk/ascend-sdk-python/ascend/client.py:576
with AscendClient(api_key="owkai_xxx") as client:
    result = client.submit_action(action)
    if result.is_approved():
        execute_action()
# Session is automatically closed

Connection Testing

# Source: sdk/ascend-sdk-python/ascend/client.py:324
# Test API connectivity
status = client.test_connection()

if status.is_connected():
    print(f"Connected to ASCEND v{status.api_version}")
    print(f"Latency: {status.latency_ms:.0f}ms")
else:
    print(f"Connection failed: {status.error}")

Advanced Configuration

Custom Headers

# The SDK automatically adds these headers:
# - Authorization: Bearer <api_key>
# - X-API-Key: <api_key>
# - User-Agent: ascend-sdk/1.0.0 Python
# - Content-Type: application/json
# - X-Correlation-ID: <unique_id>

Retry Behavior

The SDK automatically retries failed requests with exponential backoff:

Max retries: 3
Backoff factor: 0.5 seconds
Retried status codes: 429, 500, 502, 503, 504

Debug Mode

# Enable debug logging
import logging
logging.basicConfig(level=logging.DEBUG)

client = AscendClient(
    api_key="owkai_xxx",
    debug=True
)

# All requests/responses are logged (API keys masked)

Complete Example

#!/usr/bin/env python3
"""
Production-Ready ASCEND Integration
"""
import os
import sys
import logging
from ascend import AscendClient, AgentAction
from ascend.exceptions import (
    AuthenticationError,
    NetworkError,
    TimeoutError
)

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def create_client() -> AscendClient:
    """Initialize ASCEND client with error handling."""
    api_key = os.environ.get("ASCEND_API_KEY")
    if not api_key:
        logger.error("ASCEND_API_KEY environment variable not set")
        sys.exit(1)

    return AscendClient(
        api_key=api_key,
        timeout=30,
        debug=os.environ.get("ASCEND_DEBUG", "").lower() == "true"
    )

def process_customer_data(customer_id: str) -> dict:
    """Example business logic requiring governance."""
    client = create_client()

    # Test connection first
    status = client.test_connection()
    if not status.is_connected():
        logger.error(f"Cannot connect to ASCEND: {status.error}")
        raise ConnectionError("ASCEND unavailable")

    # Create governance action
    action = AgentAction(
        agent_id="data-processor-prod",
        agent_name="Customer Data Processor",
        action_type="database_read",
        resource=f"Read customer {customer_id} profile",
        tool_name="postgresql",
        resource_id=customer_id,
        action_details={
            "table": "customers",
            "columns": ["id", "name", "email"],
            "data_classification": "pii"
        },
        context={
            "environment": "production",
            "reason": "Support ticket resolution"
        }
    )

    try:
        # Submit for governance
        result = client.submit_action(action)

        if result.is_approved():
            logger.info(f"Action approved: {result.action_id}")
            # Execute actual business logic here
            return fetch_customer(customer_id)

        elif result.is_denied():
            logger.warning(f"Action denied: {result.reason}")
            raise PermissionError(f"Access denied: {result.reason}")

        elif result.is_pending():
            logger.info(f"Awaiting approval: {result.action_id}")
            # Wait for decision
            final = client.wait_for_decision(
                result.action_id,
                timeout_ms=60000
            )

            if final.is_approved():
                return fetch_customer(customer_id)
            else:
                raise PermissionError("Approval not granted")

    except AuthenticationError:
        logger.error("Invalid ASCEND API key")
        raise
    except NetworkError as e:
        logger.error(f"Network error: {e}")
        # Implement fallback or fail safely
        raise
    except TimeoutError:
        logger.error("ASCEND request timed out")
        raise

def fetch_customer(customer_id: str) -> dict:
    """Placeholder for actual data fetch."""
    return {"id": customer_id, "name": "John Doe"}

if __name__ == "__main__":
    customer = process_customer_data("cust_123")
    print(f"Customer: {customer}")

Troubleshooting

Issue	Cause	Solution
`AuthenticationError`	Invalid API key	Verify `ASCEND_API_KEY`
`ValidationError: tool_name required`	Missing required field	Add `tool_name` to action
`NetworkError`	Cannot reach API	Check firewall/proxy settings
`TimeoutError`	Slow response	Increase `timeout` parameter
`RateLimitError`	Too many requests	Implement backoff/queuing

Next Steps

REST API — Direct HTTP API reference
Boto3 Wrapper — AWS SDK governance
LangChain Integration — LangChain tools

Document Version: 1.0.0 | Last Updated: December 2025

Overview​

Installation​

Requirements​

Verify Installation​

Quick Start​

Client Configuration​

Constructor Options​

Environment Variables​

AgentAction Model​

Required Fields​

All Fields​

Standard Action Types​

ActionResult Response​

Response Fields​

Helper Methods​

Waiting for Decisions​

Synchronous Wait​

Check Status Later​

Get Full Action Details​

Listing Actions​

Error Handling​

Exception Types​

Production Error Handling​

Context Manager​

Connection Testing​

Advanced Configuration​

Custom Headers​

Retry Behavior​

Debug Mode​

Complete Example​

Troubleshooting​

Next Steps​