Boto3 Wrapper Quickstart

Add ASCEND governance to your AWS SDK (Boto3) operations for automatic risk classification and authorization.

Overview

The ASCEND Boto3 wrapper automatically intercepts AWS API calls and evaluates them against your organization's policies before execution. This enables governance of AI agents that interact with AWS resources.

Installation

pip install ascend-boto3-wrapper

Or install alongside the main SDK:

pip install ascend-ai-sdk ascend-boto3-wrapper

Quick Start

1. Initialize the Governed Client

from ascend_boto3 import GovernedBoto3Session
from ascend import AscendClient, FailMode

# Initialize ASCEND client
ascend = AscendClient(
    api_key="owkai_your_api_key",
    agent_id="aws-automation-agent",
    agent_name="AWS Automation Bot",
    fail_mode=FailMode.CLOSED
)

# Create governed Boto3 session
session = GovernedBoto3Session(
    ascend_client=ascend,
    region_name="us-east-1"
)

# Create governed service clients
s3 = session.client("s3")
dynamodb = session.client("dynamodb")

2. Use AWS Services with Automatic Governance

# S3 operations are automatically governed
response = s3.list_buckets()  # Evaluated as "aws.s3.list_buckets"

# DynamoDB operations
response = dynamodb.put_item(
    TableName="users",
    Item={"id": {"S": "123"}, "name": {"S": "John"}}
)  # Evaluated as "aws.dynamodb.put_item"

3. Handle Authorization Results

from ascend_boto3 import GovernanceError, PendingApprovalError

try:
    # High-risk operation
    s3.delete_bucket(Bucket="production-data")

except PendingApprovalError as e:
    print(f"Operation requires approval: {e.approval_id}")
    print(f"Required approvers: {e.approvers}")

except GovernanceError as e:
    print(f"Operation denied: {e.reason}")
    print(f"Risk score: {e.risk_score}")

Risk Classification

The Boto3 wrapper automatically classifies AWS operations by risk level:

Risk Level	Example Operations
Low	list_buckets, describe_instances, get_item
Medium	put_object, create_instance, put_item
High	delete_bucket, terminate_instances, delete_table
Critical	delete_bucket_policy, IAM modifications, KMS key deletion

Configuration

Session Options

session = GovernedBoto3Session(
    ascend_client=ascend,
    region_name="us-east-1",

    # Governance options
    fail_mode="closed",           # "closed" (block) or "open" (allow)
    log_all_operations=True,      # Log all AWS calls to ASCEND
    skip_read_operations=False,   # Skip governance for read-only ops

    # Risk overrides
    custom_risk_mappings={
        "s3:DeleteObject": "high",
        "lambda:InvokeFunction": "medium"
    },

    # Excluded operations (bypass governance)
    excluded_operations=[
        "sts:GetCallerIdentity",
        "s3:HeadBucket"
    ]
)

Environment Variables

Variable	Description	Default
ASCEND_BOTO3_FAIL_MODE	Fail mode ("closed" or "open")	closed
ASCEND_BOTO3_LOG_READS	Log read operations	false
ASCEND_BOTO3_SKIP_GOVN	Skip governance entirely	false

Complete Example

"""
AWS S3 Data Pipeline with ASCEND Governance
"""
from ascend_boto3 import GovernedBoto3Session, PendingApprovalError, GovernanceError
from ascend import AscendClient, FailMode
import json

# Initialize clients
ascend = AscendClient(
    api_key="owkai_prod_xxxxxxxxxxxx",
    agent_id="data-pipeline-agent",
    agent_name="S3 Data Pipeline",
    environment="production",
    fail_mode=FailMode.CLOSED
)

session = GovernedBoto3Session(
    ascend_client=ascend,
    region_name="us-east-1",
    log_all_operations=True
)

s3 = session.client("s3")
dynamodb = session.client("dynamodb")


def process_data_pipeline(source_bucket: str, dest_bucket: str, key: str) -> dict:
    """Process data with governed AWS operations."""

    result = {
        "status": "success",
        "steps": []
    }

    # Step 1: Read source data (low risk)
    try:
        response = s3.get_object(Bucket=source_bucket, Key=key)
        data = json.loads(response["Body"].read().decode("utf-8"))
        result["steps"].append({"step": "read", "status": "success"})

    except GovernanceError as e:
        return {
            "status": "denied",
            "step": "read",
            "reason": e.reason
        }

    # Step 2: Process and write to destination (medium risk)
    processed_data = transform_data(data)

    try:
        s3.put_object(
            Bucket=dest_bucket,
            Key=f"processed/{key}",
            Body=json.dumps(processed_data),
            ContentType="application/json"
        )
        result["steps"].append({"step": "write", "status": "success"})

    except PendingApprovalError as e:
        return {
            "status": "pending_approval",
            "step": "write",
            "approval_id": e.approval_id,
            "message": "Write operation requires approval"
        }

    except GovernanceError as e:
        return {
            "status": "denied",
            "step": "write",
            "reason": e.reason
        }

    # Step 3: Update DynamoDB tracking (medium risk)
    try:
        dynamodb.put_item(
            TableName="pipeline_runs",
            Item={
                "run_id": {"S": f"run_{key}"},
                "source": {"S": source_bucket},
                "destination": {"S": dest_bucket},
                "status": {"S": "completed"},
                "records": {"N": str(len(processed_data))}
            }
        )
        result["steps"].append({"step": "tracking", "status": "success"})

    except GovernanceError as e:
        # Non-critical - log but continue
        result["steps"].append({
            "step": "tracking",
            "status": "skipped",
            "reason": e.reason
        })

    return result


def transform_data(data: dict) -> dict:
    """Transform data (placeholder)."""
    return {"processed": True, "original": data}


# Run pipeline
if __name__ == "__main__":
    result = process_data_pipeline(
        source_bucket="raw-data-bucket",
        dest_bucket="processed-data-bucket",
        key="data/2024/01/15/events.json"
    )
    print(json.dumps(result, indent=2))

Service-Specific Patterns

S3 Operations

s3 = session.client("s3")

# Read operations (low risk)
buckets = s3.list_buckets()
objects = s3.list_objects_v2(Bucket="my-bucket")
data = s3.get_object(Bucket="my-bucket", Key="file.txt")

# Write operations (medium risk)
s3.put_object(Bucket="my-bucket", Key="new-file.txt", Body="data")
s3.copy_object(CopySource={"Bucket": "src", "Key": "key"}, Bucket="dst", Key="key")

# Delete operations (high risk)
s3.delete_object(Bucket="my-bucket", Key="file.txt")
s3.delete_bucket(Bucket="old-bucket")  # May require approval

DynamoDB Operations

dynamodb = session.client("dynamodb")

# Read operations (low risk)
item = dynamodb.get_item(TableName="users", Key={"id": {"S": "123"}})
items = dynamodb.query(
    TableName="users",
    KeyConditionExpression="id = :id",
    ExpressionAttributeValues={":id": {"S": "123"}}
)

# Write operations (medium risk)
dynamodb.put_item(TableName="users", Item={"id": {"S": "456"}, "name": {"S": "Jane"}})
dynamodb.update_item(
    TableName="users",
    Key={"id": {"S": "123"}},
    UpdateExpression="SET #n = :name",
    ExpressionAttributeNames={"#n": "name"},
    ExpressionAttributeValues={":name": {"S": "Updated"}}
)

# Delete operations (high risk)
dynamodb.delete_item(TableName="users", Key={"id": {"S": "123"}})
dynamodb.delete_table(TableName="old_table")  # May require approval

Lambda Operations

lambda_client = session.client("lambda")

# Invoke (medium risk)
response = lambda_client.invoke(
    FunctionName="my-function",
    Payload=json.dumps({"key": "value"})
)

# Configuration changes (high risk)
lambda_client.update_function_configuration(
    FunctionName="my-function",
    MemorySize=512
)

Error Handling

from ascend_boto3 import (
    GovernanceError,
    PendingApprovalError,
    ServiceUnavailableError
)
from botocore.exceptions import ClientError

try:
    s3.delete_bucket(Bucket="important-bucket")

except PendingApprovalError as e:
    # Operation requires human approval
    print(f"Approval required: {e.approval_id}")
    print(f"Approvers: {e.approvers}")
    # Store approval ID for later checking

except GovernanceError as e:
    # Operation denied by policy
    print(f"Denied: {e.reason}")
    print(f"Violations: {e.policy_violations}")
    print(f"Risk score: {e.risk_score}")

except ServiceUnavailableError as e:
    # ASCEND service unavailable
    if e.fail_mode == "open":
        print("Warning: Governance bypassed due to service unavailability")
    else:
        print("Error: Operation blocked - governance service unavailable")

except ClientError as e:
    # Standard AWS error
    print(f"AWS error: {e.response['Error']['Message']}")

Monitoring and Logging

# Enable detailed logging
import logging

logging.basicConfig(level=logging.DEBUG)
logging.getLogger("ascend_boto3").setLevel(logging.DEBUG)

# Custom metrics callback
def metrics_callback(operation: str, service: str, risk: str, allowed: bool, duration_ms: int):
    print(f"[Metrics] {service}:{operation} risk={risk} allowed={allowed} duration={duration_ms}ms")

session = GovernedBoto3Session(
    ascend_client=ascend,
    region_name="us-east-1",
    metrics_callback=metrics_callback
)

Next Steps

• API Reference - Complete method documentation
• Lambda Authorizer - Server-side AWS governance

Support

• Documentation: https://docs.owkai.app
• Support: support@owkai.app
• GitHub Issues: https://github.com/owkai/ascend-boto3-wrapper/issues