Skip to main content

Python SDK

FieldValue
Document IDASCEND-SDK-011
Version2026.04
Last UpdatedApril 2026
AuthorAscend Engineering Team
PublisherOW-KAI Technologies Inc.
ClassificationEnterprise Client Documentation
ComplianceSOC 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-ai-sdk) provides enterprise-grade governance integration for Python-based AI agents. It features automatic retry, circuit breaker patterns, and comprehensive error handling.

API Key Security

Store API keys in environment variables, never in source code or version control. Leaked keys grant full governance access to your organization's agents and actions.

Installation​

pip install ascend-ai-sdk==2.5.2

Package Info: ascend-ai-sdk on PyPI | Version: 2.5.2 | 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__}")

SDK Flow​

Quick Start​

# Source: ascend/client.py
import os
from ascend import AscendClient, AgentAction

# Initialize client
client = AscendClient(
    api_key=os.environ["ASCEND_API_KEY"],
    api_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: ascend/client.py
client = AscendClient(
    api_key="owkai_xxx",              # Required: Your API key
    api_url="https://pilot.owkai.app", # API endpoint
    timeout=30,                        # Request timeout (seconds)
    debug=False                        # Enable debug logging
)
ParameterTypeDefaultDescription
api_keystrRequiredASCEND API key (or ASCEND_API_KEY env var)
api_urlstrhttps://pilot.owkai.appAPI endpoint URL
timeoutint30Request timeout in seconds
debugboolFalseEnable 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: ascend/models.py
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​

FieldTypeRequiredDescription
agent_idstrYesUnique agent identifier
agent_namestrYesHuman-readable agent name
action_typestrYesAction category (see below)
resourcestrYesDescription of the action
tool_namestrYesName of tool/service being used
resource_idstrNoTarget resource identifier
action_detailsdictNoAction-specific parameters
contextdictNoExecution context
risk_indicatorsdictNoPre-computed risk signals

Standard Action Types​

CategoryAction Types
Databasedatabase_read, database_write, database_delete, database_schema
File Systemfile_read, file_write, file_delete, file_execute
Networkhttp_request, api_call, email_send, webhook_trigger
Systemprocess_spawn, config_change, credential_access
AImodel_inference, prompt_injection, data_extraction
Financialtransaction, payment_process, refund

ActionResult Response​

Response Fields​

# Source: ascend/models.py
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}")
FieldTypeDescription
action_idstrUnique identifier for this action
statusstrCurrent status: approved, denied, pending
decisionstrSame as status (for compatibility)
risk_scorefloatRisk score (0-100)
risk_levelstrRisk level: low, medium, high, critical
reasonstrExplanation of decision
policy_matchedstrName of matched policy
timestampstrISO 8601 timestamp
metadatadictAdditional 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: ascend/client.py
# 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: ascend/client.py
# Store action ID
action_id = result.action_id

# Continue with your application logic

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

Get Full Action Details​

# Source: ascend/client.py
# 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: ascend/client.py
# 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)

SDK 2.5.x — Capabilities​

Version 2.3.0 adds four net-new methods plus orchestration parameters on evaluate_action. All additions are backwards-compatible — existing callers are unaffected.

Multi-agent orchestration (FEAT-007)​

Link related agent actions into a single orchestration session so the backend can propagate risk, enforce cross-tenant scoping on parent actions, and audit the full delegation trail.

# Source: ascend/client.py
root = client.evaluate_action(
    action_type="orchestrator.plan",
    resource="customer-onboarding-flow",
    orchestration_session_id="sess-onboard-7f3a",
    orchestration_depth=0,   # root of the session
    wait_for_decision=False,
)

# Any subsequent delegated action links back via parent_action_id + session
worker = client.evaluate_action(
    action_type="database.write",
    resource="customers.create",
    orchestration_session_id="sess-onboard-7f3a",
    parent_action_id=int(root.action_id),
    orchestration_depth=1,
    wait_for_decision=False,
)

Server-enforced security:

ConditionResponse
parent_action_id does not exist in caller's organizationHTTP 403, immutable audit event ORCHESTRATION_FIELD_REJECTED
orchestration_session_id does not match parent_action.orchestration_session_id (graft attempt)HTTP 403 (CRITICAL), immutable audit event ORCHESTRATION_GRAFT_ATTEMPT
orchestration_depth outside 0..5HTTP 422
Any DB error during validationHTTP 403 — fail-secure, never silently drops

Client-side, depth is also rejected locally before the network call (ValidationError).

See Multi-Agent Orchestration for the full security model.

Wire a registered DeployedModel to an agent. The backend validates that the model belongs to the caller's organization and is in an approved or partially_approved compliance state.

response = client.link_model_to_agent(
    agent_id="financial-advisor-prod",
    model_id=42,          # DeployedModel.id
)
# response = {"success": True, "agent": {...}, "message": "..."}

See AI Model Registry.

Register a supply-chain component (FEAT-005)​

Record an external AI component (library, model, dataset, framework, tool, or service) your organization depends on. This feeds the AI supply-chain risk surface and impact analysis.

response = client.register_supply_chain_component(
    component_id="hf-bert-base",
    component_name="bert-base-uncased",
    component_type="model",          # matches backend ComponentType enum
    provider="HuggingFace",
    version="1.0.0",
    license_type="Apache-2.0",
    source_url="https://huggingface.co/bert-base-uncased",
    provenance_verified=True,
    risk_level="medium",
    package_name="transformers",
    package_ecosystem="pypi",
)
# response = {"success": True, "component": {...}, "message": "..."}

The endpoint POST /api/v1/supply-chain/components accepts either JWT admin or admin-role API key (SEC-096 dual auth). See AI Supply Chain.

Kill-switch HTTP fallback (SEC-103)​

The primary kill-switch channel is SNS/SQS. For environments where the SDK can't hold AWS credentials or SQS is unreachable, 2.3.0 adds HTTP fallback polling + acknowledgement:

# Poll for pending BLOCK/UNBLOCK/SUSPEND/RATE_LIMIT/QUARANTINE commands.
# Returns commands targeted at this agent AND org-broadcast commands.
commands = client.get_pending_commands(agent_id="agent-001")

for cmd in commands:
    cmd_id = cmd["command_id"]
    cmd_type = cmd["command_type"]      # "BLOCK" | "UNBLOCK" | ...
    reason = cmd["reason"]

    if cmd_type == "BLOCK":
        # Halt agent operations
        ...

    # Ack as received — receipt-only; off-tenant acks are 403 fail-secure
    client.ack_command(command_id=cmd_id, agent_id="agent-001")

See Kill Switch — HTTP fallback for the full polling pattern and the differences from the boto3 SQS consumer.

MCP governance (shipped in 2.2.0, canonical reference here)​

evaluate_mcp_action posts to the dedicated MCP pipeline at POST /api/v1/mcp/actions/submit — not the legacy /api/mcp-governance/* paths (retired).

decision = client.evaluate_mcp_action(
    mcp_server="filesystem-server",
    namespace="filesystem",
    verb="read_file",
    resource="/data/reports/financial.csv",
    # action_type defaults to "<namespace>.<verb>"
    parameters={"encoding": "utf-8"},
    context={"session_id": "sess_abc"},
    environment="production",
)

MCPKillSwitchConsumer (also in ascend.mcp) lets your SQS poll loop update in-process block state without an AWS dependency:

from ascend.mcp import MCPKillSwitchConsumer

consumer = MCPKillSwitchConsumer()

# In your SQS poll loop — backend publishes target_type="mcp_server" messages
for msg in sqs_messages:
    consumer.apply_message(msg)   # fail-secure on malformed

# In the governed tool, check before executing:
if consumer.is_blocked("filesystem-server"):
    raise RuntimeError("MCP server blocked by administrator")

SDK 2.5.1 — Model & MCP Governance​

Pass model_id to trigger model registry compliance enforcement (G-P0-02):

result = client.evaluate_action(
    action_type="model_inference",
    resource="ml_pipeline",
    model_id="gpt-4-production",
    wait_for_decision=False,
)
# registry_checked: True
# compliance_status: approved | partially_approved | denied
# enforcement: SR-11-7/EU-AI-ACT-ART9
print(result.model_governance)

Pass mcp_server_name to trigger Layer 13 MCP governance enforcement (G-P0-01):

result = client.evaluate_action(
    action_type="tool_call",
    resource="crm_system",
    mcp_server_name="salesforce-mcp",
    wait_for_decision=False,
)
# server_registered: True
# enforcement: Layer 13
print(result.mcp_governance)

Non-compliant models and unregistered MCP servers return HTTP 403.

Error Handling​

Exception Types​

# Source: 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: ascend/client.py
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: ascend/client.py
# Test API connectivity
status = client.test_connection()

if status.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/2.5.2 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.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:
    """Fetch customer record from your database."""
    return {"id": customer_id, "name": "John Doe"}

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

Troubleshooting​

AuthenticationError on client initialization​

Cause: The API key is missing, empty, or incorrectly formatted. The SDK validates the key format during initialization. Solution: Verify your API key starts with ask_ or owkai_ and is at least 16 characters. Set it via environment variable: export ASCEND_API_KEY=ask_your_key_here or pass directly: AscendClient(api_key="ask_...").

ConnectionError: Failed to connect to ASCEND API​

Cause: The SDK cannot reach the ASCEND API endpoint. This could be a DNS resolution failure, firewall rule, or incorrect base URL. Solution: Test connectivity: curl https://pilot.owkai.app/health. If using a custom endpoint, verify api_url is correct. Check proxy settings if behind a corporate firewall.

Actions stuck in pending_approval status​

Cause: The action requires human approval based on its risk score, but no approver has acted on it yet. The wait_for_decision() method polls until a decision is made or timeout is reached. Solution: Check the ASCEND dashboard for pending approvals. Verify your organization has approvers configured for the required risk level. Increase the timeout_ms parameter if approvals take longer than expected.

RateLimitError: 429 Too Many Requests​

Cause: Your API key has exceeded its rate limit for the current window. Solution: Check the Retry-After header in the error response for the wait duration. Implement exponential backoff in your retry logic. Contact your administrator to increase rate limits if needed.

Next Steps​

Document Version: 2026.04 | Last Updated: April 2026