Python SDK Quickstart
Get your AI agent governed by ASCEND in under 5 minutes.
Prerequisites
- Python 3.8 or higher
- ASCEND API key (obtain from the Console)
- pip package manager
Installation
pip install ascend-ai-sdk
Quick Start
1. Initialize the Client
from ascend import AscendClient, FailMode
# Initialize with fail-closed mode (recommended for production)
client = AscendClient(
api_key="owkai_your_api_key_here",
agent_id="my-agent-001",
agent_name="My AI Agent",
environment="production",
fail_mode=FailMode.CLOSED
)
2. Register Your Agent
# Register agent with ASCEND platform
registration = client.register(
agent_type="automation",
capabilities=["database.query", "file.read", "api.call"],
allowed_resources=["production_db", "customer_api"]
)
print(f"Agent registered with trust level: {registration.get('trust_level')}")
3. Evaluate Actions Before Execution
from ascend import Decision
# Request authorization for an action
decision = client.evaluate_action(
action_type="database.query",
resource="production_db",
parameters={"query": "SELECT * FROM customers WHERE id = 123"}
)
# Check the decision
if decision.decision == Decision.ALLOWED:
# Execute the action
result = execute_database_query()
# Log successful completion
client.log_action_completed(
action_id=decision.action_id,
result={"rows_returned": len(result)},
duration_ms=150
)
elif decision.decision == Decision.PENDING:
# Action requires human approval
print(f"Awaiting approval: {decision.approval_request_id}")
print(f"Required approvers: {decision.required_approvers}")
elif decision.decision == Decision.DENIED:
# Action was denied
print(f"Action denied: {decision.reason}")
print(f"Policy violations: {decision.policy_violations}")
Environment Variables
The SDK supports configuration via environment variables:
| Variable | Description | Required |
|---|---|---|
ASCEND_API_KEY | Your organization API key | Yes |
ASCEND_API_URL | API endpoint (default: https://pilot.owkai.app) | No |
ASCEND_AGENT_ID | Default agent identifier | No |
ASCEND_AGENT_NAME | Default agent name | No |
import os
os.environ["ASCEND_API_KEY"] = "owkai_your_api_key"
os.environ["ASCEND_AGENT_ID"] = "my-agent-001"
# Client will use environment variables
client = AscendClient(agent_name="My AI Agent")
Complete Example
"""
Complete example: A customer service agent that processes refunds
"""
from ascend import AscendClient, FailMode, Decision
import time
# Initialize client
client = AscendClient(
api_key="owkai_prod_xxxxxxxxxxxx",
agent_id="customer-service-agent",
agent_name="Customer Service Bot",
environment="production",
fail_mode=FailMode.CLOSED,
timeout=5
)
# Register the agent
client.register(
agent_type="automation",
capabilities=["transaction.refund", "customer.lookup"],
allowed_resources=["stripe_api", "customer_db"]
)
def process_refund(customer_id: str, amount: float, reason: str) -> dict:
"""Process a customer refund with ASCEND governance."""
# Request authorization
decision = client.evaluate_action(
action_type="transaction.refund",
resource="stripe_api",
parameters={
"customer_id": customer_id,
"amount": amount,
"currency": "USD",
"reason": reason
},
context={
"session_id": "sess_abc123",
"user_request": f"Refund ${amount} for customer {customer_id}"
}
)
# Handle decision
if decision.decision == Decision.ALLOWED:
start_time = time.time()
try:
# Execute the refund
result = {
"refund_id": "ref_123456",
"status": "processed",
"amount": amount
}
# Log completion
duration_ms = int((time.time() - start_time) * 1000)
client.log_action_completed(
action_id=decision.action_id,
result=result,
duration_ms=duration_ms
)
return result
except Exception as e:
# Log failure
client.log_action_failed(
action_id=decision.action_id,
error={"code": "EXECUTION_ERROR", "message": str(e)}
)
raise
elif decision.decision == Decision.PENDING:
return {
"status": "pending_approval",
"approval_id": decision.approval_request_id,
"message": "Refund requires manager approval"
}
else:
return {
"status": "denied",
"reason": decision.reason,
"risk_score": decision.risk_score
}
# Execute
result = process_refund(
customer_id="cust_123",
amount=150.00,
reason="Product defect"
)
print(result)
Error Handling
from ascend import (
AscendClient,
AuthenticationError,
AuthorizationError,
TimeoutError,
RateLimitError,
CircuitBreakerOpen
)
try:
decision = client.evaluate_action(
action_type="database.query",
resource="production_db",
parameters={"query": "SELECT * FROM users"}
)
except AuthenticationError as e:
# Invalid or expired API key
print(f"Authentication failed: {e}")
except AuthorizationError as e:
# Action explicitly denied by policy
print(f"Authorization denied: {e}")
print(f"Policy violations: {e.policy_violations}")
print(f"Risk score: {e.risk_score}")
except TimeoutError as e:
# Request timed out
print(f"Request timed out after {e.timeout_seconds}s")
except RateLimitError as e:
# Rate limit exceeded
print(f"Rate limited. Retry after {e.retry_after} seconds")
except CircuitBreakerOpen as e:
# Service appears to be down
print(f"Service unavailable. Recovery in {e.recovery_time}s")
Fail Mode Configuration
The SDK supports two fail modes:
| Mode | Behavior | Use Case |
|---|---|---|
CLOSED | Block actions when ASCEND is unreachable | Production (recommended) |
OPEN | Allow actions when ASCEND is unreachable | Development/testing |
from ascend import AscendClient, FailMode
# Production: fail-closed (secure)
prod_client = AscendClient(
api_key="owkai_prod_xxx",
agent_id="prod-agent",
agent_name="Production Agent",
fail_mode=FailMode.CLOSED # Default
)
# Development: fail-open (for testing)
dev_client = AscendClient(
api_key="owkai_dev_xxx",
agent_id="dev-agent",
agent_name="Dev Agent",
fail_mode=FailMode.OPEN
)
Next Steps
- API Reference - Complete method documentation
- MCP Integration - Integrate with MCP servers
- Examples - Production code patterns
Support
- Documentation: https://docs.owkai.app
- Support: support@owkai.app
- GitHub Issues: https://github.com/owkai/ascend-sdk-python/issues