API Reference - Bedrock Budgeteer

Overview

This document provides comprehensive API reference for all constructs, classes, and interfaces in the Bedrock Budgeteer system. This reference is intended for developers extending or maintaining the system.

CDK Constructs

BedrockBudgeteerStack

Main application stack that orchestrates all system components.

Constructor:

BedrockBudgeteerStack(
    scope: Construct,
    construct_id: str,
    environment_name: str = "production",
    kms_key: Optional[kms.IKey] = None,
    **kwargs
)

Parameters:

  • scope: CDK construct scope
  • construct_id: Unique identifier for the stack
  • environment_name: Environment name (default: “production”)
  • kms_key: Optional customer-managed KMS key for encryption

Properties:

@property
def dynamodb_tables(self) -> Dict[str, dynamodb.Table]
    """Expose DynamoDB tables for use by other constructs"""

@property 
def iam_roles(self) -> Dict[str, iam.Role]
    """Expose IAM roles for use by other constructs"""

@property
def sns_topics(self) -> Dict[str, sns.Topic]
    """Expose SNS topics for use by other constructs"""

@property
def lambda_functions(self) -> Dict[str, lambda_.Function]
    """Expose Lambda functions for use by other constructs"""

@property
def step_functions_state_machines(self) -> Dict[str, sfn.StateMachine]
    """Expose Step Functions state machines"""

Usage Example:

from app.app_stack import BedrockBudgeteerStack

stack = BedrockBudgeteerStack(
    app, "BedrockBudgeteer",
    environment_name="production",
    env=cdk.Environment(account="123456789012", region="us-east-1")
)

DataStorageConstruct

Manages DynamoDB tables and data storage resources.

Constructor:

DataStorageConstruct(
    scope: Construct,
    construct_id: str,
    environment_name: str,
    kms_key: Optional[kms.IKey] = None,
    **kwargs
)

Tables Created:

  • user_budgets: User budget tracking table
  • usage_tracking: Usage and cost tracking table
  • audit_logs: System audit trail table
  • pricing: Bedrock model pricing cache table

Properties:

@property
def tables(self) -> Dict[str, dynamodb.Table]
    """Dictionary of created DynamoDB tables"""

Table Schemas:

UserBudgets Table:

{
    "partition_key": "principal_id",  # String
    "attributes": {
        "budget_limit_usd": "Number",
        "spent_usd": "Number", 
        "status": "String",  # active|suspended|grace_period
        "account_type": "String",  # bedrock_api_key|user|service
        "budget_refresh_date": "String",  # ISO8601
        "grace_deadline_epoch": "Number"
    },
    "gsi": {
        "BudgetStatusIndex": {
            "partition_key": "budget_status",
            "sort_key": "created_at"
        }
    }
}

UsageTracking Table:

{
    "partition_key": "principal_id",  # String
    "sort_key": "timestamp",  # String
    "attributes": {
        "service_name": "String",
        "cost_usd": "Number",
        "token_count": "Number",
        "model_id": "String"
    },
    "gsi": {
        "ServiceUsageIndex": {
            "partition_key": "service_name",
            "sort_key": "timestamp"
        }
    }
}

SecurityConstruct

Manages IAM roles, policies, and security resources.

Constructor:

SecurityConstruct(
    scope: Construct,
    construct_id: str,
    environment_name: str,
    **kwargs
)

Roles Created:

  • lambda_execution: Execution role for Lambda functions
  • step_functions: Execution role for Step Functions
  • eventbridge: Service role for EventBridge
  • bedrock_logging: Role for Bedrock invocation logging

Properties:

@property
def roles(self) -> Dict[str, iam.Role]
    """Dictionary of created IAM roles"""

@property
def policies(self) -> Dict[str, iam.ManagedPolicy] 
    """Dictionary of created managed policies"""

@property
def bedrock_logging_role(self) -> iam.Role
    """Bedrock logging role for console configuration"""

@property
def lambda_execution_role(self) -> iam.Role
    """Lambda execution role for additional permissions"""

Methods:

def add_bedrock_permissions(self) -> None
    """Add Bedrock API permissions for cost monitoring"""

def add_pricing_api_permissions(self) -> None
    """Add AWS Pricing API permissions for cost calculation"""

def add_cloudtrail_permissions(self) -> None
    """Add CloudTrail permissions for usage tracking"""

def add_ssm_permissions(self) -> None
    """Add SSM Parameter Store permissions"""

def add_dynamodb_permissions(self, tables: Dict[str, Any]) -> None
    """Add DynamoDB permissions for Lambda functions"""

def create_policy_template(self, service_name: str, actions: List[str], 
                          resources: List[str]) -> iam.ManagedPolicy
    """Create a reusable policy template for any AWS service"""

EventIngestionConstruct

Manages CloudTrail, EventBridge, and Kinesis Data Firehose for event collection.

Constructor:

EventIngestionConstruct(
    scope: Construct,
    construct_id: str,
    environment_name: str,
    s3_bucket: Optional[s3.Bucket] = None,
    kms_key: Optional[kms.IKey] = None,
    usage_calculator_function: Optional[lambda_.Function] = None,
    **kwargs
)

Properties:

@property
def cloudtrail_trails(self) -> Dict[str, cloudtrail.Trail]
    """Dictionary of CloudTrail trails"""

@property
def eventbridge_rules(self) -> Dict[str, events.Rule]
    """Dictionary of EventBridge rules"""

@property 
def firehose_streams(self) -> Dict[str, firehose.DeliveryStream]
    """Dictionary of Kinesis Firehose streams"""

@property
def log_groups(self) -> Dict[str, logs.LogGroup]
    """Dictionary of CloudWatch log groups"""

@property
def bedrock_invocation_log_group(self) -> logs.LogGroup
    """Bedrock invocation log group for configuration"""

Methods:

def add_eventbridge_target(self, rule_name: str, target) -> None
    """Add a target to an existing EventBridge rule"""

def configure_usage_calculator(self, usage_calculator_function: lambda_.Function) -> None
    """Configure usage calculator Lambda for real-time processing"""

CoreProcessingConstruct

Implements Lambda functions for processing events, calculating costs, and monitoring budgets.

Constructor:

CoreProcessingConstruct(
    scope: Construct,
    construct_id: str,
    environment_name: str,
    dynamodb_tables: Dict[str, dynamodb.Table],
    s3_bucket: s3.Bucket,
    lambda_execution_role: iam.Role,
    kms_key: Optional[kms.Key] = None,
    **kwargs
)

Properties:

@property
def functions(self) -> Dict[str, lambda_.Function]
    """Dictionary of Lambda functions"""

@property
def dead_letter_queues(self) -> Dict[str, sqs.Queue]
    """Dictionary of DLQ queues for monitoring"""

@property
def execution_role(self) -> iam.Role
    """Execution role for additional permissions"""

Lambda Functions Created:

  • user_setup: Initialize budgets from CloudTrail events
  • usage_calculator: Transform Bedrock logs into cost records
  • budget_monitor: Evaluate thresholds and trigger workflows
  • budget_refresh: Reset budgets and restore users
  • audit_logger: Process audit events
  • state_reconciliation: Verify IAM/DynamoDB consistency
  • pricing_manager: Manage Bedrock pricing data

WorkflowOrchestrationConstruct

Coordinates suspension and restoration workflows using Step Functions.

Constructor:

WorkflowOrchestrationConstruct(
    scope: Construct,
    construct_id: str,
    environment_name: str,
    dynamodb_tables: Dict[str, dynamodb.Table],
    lambda_functions: Dict[str, lambda_.Function],
    step_functions_role: iam.Role,
    lambda_execution_role: iam.Role,
    sns_topics: Optional[Dict[str, sns.Topic]] = None,
    kms_key: Optional[kms.Key] = None,
    **kwargs
)

Properties:

@property
def suspension_state_machine(self) -> sfn.StateMachine
    """Suspension workflow state machine"""

@property
def restoration_state_machine(self) -> sfn.StateMachine
    """Restoration workflow state machine"""

@property
def workflow_functions(self) -> Dict[str, lambda_.Function]
    """Dictionary of workflow Lambda functions"""

@property
def workflow_dlqs(self) -> Dict[str, sqs.Queue]
    """Dictionary of workflow DLQ queues"""

@property
def state_machines(self) -> Dict[str, sfn.StateMachine]
    """Dictionary of all state machines"""

Workflow Functions Created:

  • iam_utilities: IAM policy management utilities
  • grace_period: Grace period notifications
  • policy_backup: Policy backup operations
  • restoration_validation: Restoration eligibility validation

AgentCoreConstruct

Manages AgentCore budget tracking, enforcement, and management API. Enabled via the enable_agentcore_budgeting feature flag in cdk.json.

Constructor:

AgentCoreConstruct(
    scope: Construct,
    construct_id: str,
    environment_name: str,
    dynamodb_tables: Dict[str, dynamodb.Table],
    lambda_execution_role: iam.Role,
    step_functions_role: iam.Role,
    sns_topics: Optional[Dict[str, sns.Topic]] = None,
    kms_key: Optional[kms.Key] = None,
    **kwargs
)

Resources Created:

  • DynamoDB Table: agentcore-budgets — tracks per-agent and global budget state
  • Lambda Functions (4):
    • agentcore_usage_calculator: Processes AgentCore invocation events and calculates costs
    • agentcore_budget_monitor: Evaluates agent budget thresholds and triggers workflows
    • agentcore_budget_manager: Function URL API for budget management operations (see below)
    • agentcore_budget_refresh: Resets agent budgets on refresh cycle
  • EventBridge Rules: Routes AgentCore usage events to the usage calculator and budget monitor
  • Step Functions: Suspension and restoration workflows for AgentCore agents

Properties:

@property
def agentcore_table(self) -> dynamodb.Table
    """AgentCore budgets DynamoDB table"""

@property
def functions(self) -> Dict[str, lambda_.Function]
    """Dictionary of AgentCore Lambda functions"""

@property
def state_machines(self) -> Dict[str, sfn.StateMachine]
    """Dictionary of AgentCore Step Functions state machines"""

Budget Manager API (Function URL)

The agentcore_budget_manager Lambda is exposed via a Function URL with IAM authentication. All requests use a JSON body with an action field.

Authentication: AWS IAM (SigV4 signed requests)

Operations:

set_agent_budget

Set or update a per-agent budget.

Request:

{
  "action": "set_agent_budget",
  "runtime_id": "agent-runtime-id-123",
  "budget_limit_usd": 100.0
}

Response:

{
  "success": true,
  "data": {
    "runtime_id": "agent-runtime-id-123",
    "budget_limit_usd": 100.0,
    "spent_usd": 0.0,
    "status": "active",
    "updated_at": "2026-04-03T12:00:00Z"
  },
  "error": null
}
remove_agent_budget

Remove a per-agent budget (agent falls back to global budget).

Request:

{
  "action": "remove_agent_budget",
  "runtime_id": "agent-runtime-id-123"
}

Response:

{
  "success": true,
  "data": {
    "runtime_id": "agent-runtime-id-123",
    "removed": true
  },
  "error": null
}
set_global_budget

Set or update the global AgentCore budget limit.

Request:

{
  "action": "set_global_budget",
  "budget_limit_usd": 500.0
}

Response:

{
  "success": true,
  "data": {
    "budget_limit_usd": 500.0,
    "updated_at": "2026-04-03T12:00:00Z"
  },
  "error": null
}
get_budget_status

Retrieve the current budget status for an agent or the global budget.

Request (per-agent):

{
  "action": "get_budget_status",
  "runtime_id": "agent-runtime-id-123"
}

Request (global):

{
  "action": "get_budget_status"
}

Response:

{
  "success": true,
  "data": {
    "runtime_id": "agent-runtime-id-123",
    "budget_limit_usd": 100.0,
    "spent_usd": 42.50,
    "status": "active",
    "budget_usage_percent": 42.5
  },
  "error": null
}

Error Response (all operations):

{
  "success": false,
  "data": null,
  "error": "Invalid action or missing required fields"
}

manage_keys.py CLI

Standalone script that provisions and manages Bedrock API key IAM users directly via AWS APIs. No CDK deploy needed per key.

Prerequisites:

  • AWS credentials configured (aws configure or environment variables)
  • enable_key_provisioning: true in cdk.json and deployed (for runtime support: SSM params, IAM permissions, SNS wiring)
  • Activate CostAllocation:Team and CostAllocation:Purpose as user-defined cost allocation tags in the AWS Billing console for Cost Explorer grouping

Commands:

# Create a tagged IAM user with Bedrock access
python manage_keys.py add --team platform --purpose chatbot-prod --budget-tier medium

# List all BedrockAPIKey-* users with their tags
python manage_keys.py list

# Remove an IAM user (detaches policies, deletes access keys, deletes user)
python manage_keys.py remove --team platform --purpose chatbot-prod

Parameters (add):

  • --team (required): Team name (e.g., platform, ml-ops)
  • --purpose (required): Use case (e.g., chatbot-prod, batch-inference)
  • --budget-tier (optional): low ($1), medium ($5), or high ($25). Default: low

IAM User Created:

  • Named BedrockAPIKey-{team}-{purpose} with AmazonBedrockLimitedAccess managed policy
  • Tags Applied (7): BedrockBudgeteer:Team, BedrockBudgeteer:Purpose, BedrockBudgeteer:BudgetTier, BedrockBudgeteer:Provisioned (=script), BedrockBudgeteer:ManagedBy, CostAllocation:Team, CostAllocation:Purpose

How it works: The script creates the IAM user directly. The existing CloudTrail → EventBridge → user_setup Lambda pipeline detects the new user, reads its tags, and automatically registers the budget based on the tier.

CostAllocationReportingConstruct

Daily Cost Explorer sync and reconciliation for Bedrock cost reporting. Enabled via the enable_cost_allocation_reporting feature flag in cdk.json.

Constructor:

CostAllocationReportingConstruct(
    scope: Construct,
    construct_id: str,
    environment_name: str,
    lambda_execution_role: iam.Role,
    usage_tracking_table: dynamodb.Table,
    sns_topics: Optional[Dict[str, sns.Topic]] = None,
    kms_key: Optional[kms.Key] = None,
    **kwargs
)

Resources Created:

  • Lambda Functions (2):
    • cost_allocation_sync: Queries Cost Explorer daily, publishes CloudWatch metrics by team/purpose/tier
    • cost_reconciliation: Compares Cost Explorer totals vs internal usage_tracking table, alerts on >10% drift
  • DLQs (2): Dead letter queues for each Lambda
  • EventBridge Schedules (2): Daily at 06:00 UTC (sync) and 07:00 UTC (reconciliation)

Lambda Environment Variables:

  • ENVIRONMENT: Environment name
  • USAGE_TRACKING_TABLE: DynamoDB table name (reconciliation Lambda only)
  • OPERATIONAL_ALERTS_SNS_TOPIC_ARN: SNS topic for drift alerts (reconciliation Lambda only)

CloudWatch Metrics Published (namespace: BedrockBudgeteer/CostAllocation):

  • CostByTeam, CostByPurpose, CostByTier, TotalBedrockCost, CostReconciliationDrift

MonitoringConstruct

Manages CloudWatch resources, dashboards, and alarms.

Constructor:

MonitoringConstruct(
    scope: Construct,
    construct_id: str,
    environment_name: str,
    **kwargs
)

Properties:

@property
def topics(self) -> Dict[str, sns.Topic]
    """Dictionary of SNS topics"""

@property
def alarms(self) -> Dict[str, cloudwatch.Alarm]
    """Dictionary of CloudWatch alarms"""

@property
def log_groups(self) -> Dict[str, logs.LogGroup]
    """Dictionary of log groups"""

Methods:

def add_lambda_monitoring(self, function_name: str, lambda_function) -> None
    """Add monitoring for a Lambda function"""

def add_dynamodb_monitoring(self, table_name: str, table) -> None
    """Add monitoring for a DynamoDB table"""

def add_stepfunctions_monitoring(self, state_machine_name: str, state_machine) -> None
    """Add monitoring for Step Functions state machines"""

def add_email_subscription(self, topic_name: str, email: str) -> None
    """Add email subscription to SNS topic"""

def add_slack_subscription(self, topic_name: str, webhook_url: str) -> None
    """Add Slack webhook subscription to SNS topic"""

def create_custom_business_metrics(self) -> None
    """Create custom CloudWatch metrics for business monitoring"""

SNS Topics Created:

  • operational_alerts: System operational issues
  • budget_alerts: Budget threshold violations
  • high_severity: Critical system alerts

ConfigurationConstruct

Manages SSM Parameter Store hierarchy and application configuration.

Constructor:

ConfigurationConstruct(
    scope: Construct,
    construct_id: str,
    environment_name: str,
    kms_key: Optional[kms.Key] = None,
    **kwargs
)

Properties:

@property
def parameters(self) -> Dict[str, ssm.StringParameter]
    """Dictionary of created SSM parameters"""

Methods:

def get_parameter_reference(self, category: str, key: str) -> str
    """Get SSM parameter reference for use in other constructs"""

def create_custom_parameter(self, category: str, key: str, value: str,
                          description: str, secure: bool = False) -> ssm.StringParameter
    """Create a custom parameter in the hierarchy"""

Parameter Hierarchy:

/bedrock-budgeteer/
├── global/
│   ├── thresholds_percent_warn
│   ├── thresholds_percent_critical  
│   ├── default_user_budget_usd
│   └── grace_period_seconds
└── production/
    └── cost/
        └── budget_refresh_period_days

Workflow Classes

SuspensionWorkflow

Defines the Step Functions state machine for user suspension workflow.

Constructor:

SuspensionWorkflow(
    scope: Construct,
    environment_name: str,
    dynamodb_tables: Dict[str, dynamodb.Table],
    workflow_lambda_functions: Dict[str, lambda_.Function],
    step_functions_role: iam.Role
)

Methods:

def create_suspension_workflow(self) -> sfn.StateMachine
    """Create Step Functions state machine for user suspension workflow"""

Workflow Steps:

  1. Send Grace Notification
  2. Grace Period Wait (configurable duration)
  3. Send Final Warning
  4. Apply Full Suspension
  5. Update User Status
  6. Send Audit Event

RestorationWorkflow

Defines the Step Functions state machine for user restoration workflow.

Constructor:

RestorationWorkflow(
    scope: Construct,
    environment_name: str,
    dynamodb_tables: Dict[str, dynamodb.Table],
    workflow_lambda_functions: Dict[str, lambda_.Function],
    step_functions_role: iam.Role
)

Methods:

def create_restoration_workflow(self) -> sfn.StateMachine
    """Create Step Functions state machine for user restoration workflow"""

Workflow Steps:

  1. Validate Automatic Restoration
  2. Validation Choice
  3. Restore Access (if validated)
  4. Validate Access Restoration
  5. Reset Budget Status
  6. Send Audit Event

WorkflowBase

Base class for workflow definitions with common utilities.

Methods:

def create_lambda_invoke_task(self, task_name: str, function_key: str, 
                            input_data: Dict, result_path: str) -> sfn_tasks.LambdaInvoke
    """Create a Lambda invoke task with error handling"""

def create_dynamodb_update_task(self, task_name: str, table_key: str,
                              key: Dict, update_expression: str,
                              expression_attribute_names: Dict,
                              expression_attribute_values: Dict,
                              result_path: str) -> sfn_tasks.DynamoUpdateItem
    """Create a DynamoDB update task"""

def create_failure_state(self, state_name: str, error: str, cause: str) -> sfn.Fail
    """Create a failure state for error handling"""

Shared Utilities

ConfigurationManager

Manages SSM parameter configuration with time-based caching.

Caching: Parameters are cached for 5 minutes (TTL). After expiry, the next get_parameter call refreshes from SSM. This means SSM parameter changes (e.g., threshold updates, budget adjustments) take up to 5 minutes to take effect in running Lambda functions.

Methods:

@classmethod
def get_parameter(cls, parameter_name: str, default_value: Any = None) -> Any
    """Get parameter from SSM Parameter Store with 5-min TTL cache.
    Returns default_value if SSM lookup fails."""

@classmethod
def get_budget_thresholds(cls) -> Dict[str, float]
    """Get budget threshold configuration (warn_percent, critical_percent)"""

DynamoDBHelper

Helper functions for DynamoDB operations.

Methods:

@staticmethod
def decimal_to_float(obj) -> Any
    """Convert Decimal objects to float for JSON serialization"""

@staticmethod
def float_to_decimal(obj) -> Any
    """Convert float objects to Decimal for DynamoDB storage"""

@staticmethod
def get_user_budget(principal_id: str) -> Optional[Dict]
    """Get user budget record from DynamoDB"""

@staticmethod
def update_user_budget(principal_id: str, spent_usd: float) -> bool
    """Update user budget spent amount"""

BedrockPricingCalculator

Calculates costs for Bedrock API usage.

Methods:

@staticmethod
def calculate_cost(model_id: str, input_tokens: int, output_tokens: int, 
                  region: str = "us-east-1") -> float
    """Calculate cost for Bedrock API usage"""

@staticmethod
def fetch_pricing_from_api(model_id: str, region: str) -> Optional[Dict]
    """Fetch current pricing from AWS Pricing API"""

@staticmethod
def get_cached_pricing(model_id: str, region: str) -> Optional[Dict]
    """Get pricing from DynamoDB cache"""

MetricsPublisher

Publishes custom CloudWatch metrics.

Methods:

@staticmethod
def publish_budget_metric(metric_name: str, value: float, unit: str, 
                         dimensions: Dict[str, str]) -> None
    """Publish a budget-related metric to CloudWatch"""

@staticmethod
def publish_system_metric(metric_name: str, value: float, unit: str,
                         dimensions: Dict[str, str]) -> None
    """Publish a system-related metric to CloudWatch"""

EventPublisher

Publishes events to EventBridge.

Methods:

@staticmethod
def publish_budget_event(event_type: str, detail: Dict[str, Any]) -> None
    """Publish a budget-related event to EventBridge"""

@staticmethod
def publish_audit_event(event_type: str, detail: Dict[str, Any]) -> None
    """Publish an audit event to EventBridge"""

Lambda Function APIs

User Setup Lambda

Initializes budget entries for new users.

Event Input:

{
  "detail": {
    "eventName": "CreateUser",
    "responseElements": {
      "user": {
        "userName": "BedrockAPIKey-UserName"
      }
    }
  }
}

Response:

{
  "statusCode": 200,
  "principal_id": "BedrockAPIKey-UserName",
  "budget_initialized": true,
  "budget_limit_usd": 1.0
}

Usage Calculator Lambda

Processes Bedrock invocation logs and calculates costs.

Event Input (Firehose):

{
  "records": [
    {
      "data": "base64-encoded-log-event"
    }
  ]
}

Response:

{
  "records": [
    {
      "recordId": "string",
      "result": "Ok",
      "data": "base64-encoded-transformed-data"
    }
  ]
}

Budget Monitor Lambda

Evaluates budget thresholds and triggers suspension workflows.

Event Input (Scheduled):

{
  "source": "aws.events",
  "detail-type": "Scheduled Event"
}

Response:

{
  "statusCode": 200,
  "monitored_users": 25,
  "budget_exceeded_users": 2
}

Pricing Manager Lambda

Manages Bedrock model pricing data.

Event Input:

{
  "action": "daily_refresh"
}

Response:

{
  "statusCode": 200,
  "body": {
    "refreshed": 15,
    "total": 20,
    "failed": 5,
    "populated_by": "populated by event"
  }
}

Error Handling

Standard Error Response

All Lambda functions return standardized error responses:

{
  "statusCode": 500,
  "error": "Error description",
  "details": {
    "function": "function-name",
    "timestamp": "2024-01-01T00:00:00Z",
    "request_id": "uuid"
  }
}

Dead Letter Queue Format

Failed Lambda executions are sent to DLQ with metadata:

{
  "original_event": {},
  "error_message": "string",
  "function_name": "string", 
  "timestamp": "ISO8601",
  "attempt_count": 3
}

Environment Variables

Lambda Function Environment Variables

All Lambda functions receive these standard environment variables:

{
  "ENVIRONMENT": "production",
  "USER_BUDGETS_TABLE": "bedrock-budgeteer-production-user-budgets",
  "USAGE_TRACKING_TABLE": "bedrock-budgeteer-production-usage-tracking", 
  "AUDIT_LOGS_TABLE": "bedrock-budgeteer-production-audit-logs",
  "PRICING_TABLE": "bedrock-budgeteer-production-pricing",
  "LOGS_BUCKET": "bedrock-budgeteer-production-logs"
}

Workflow-Specific Environment Variables

Workflow Lambda functions receive additional variables:

{
  "HIGH_SEVERITY_TOPIC_ARN": "arn:aws:sns:...",
  "OPERATIONAL_ALERTS_TOPIC_ARN": "arn:aws:sns:...",
  "BUDGET_ALERTS_TOPIC_ARN": "arn:aws:sns:..."
}

Event Schemas

EventBridge Event Schemas

Budget Violation Event:

{
  "source": "bedrock-budgeteer",
  "detail-type": "Suspension Workflow Required",
  "detail": {
    "principal_id": "string",
    "reason": "grace_period_expired",
    "grace_period_seconds": 300,
    "budget_data": {
      "account_type": "bedrock_api_key",
      "budget_limit_usd": 10.0,
      "spent_usd": 12.5,
      "budget_usage_percent": 125.0
    },
    "triggered_by": "budget_monitor",
    "timestamp": "2024-01-01T00:00:00Z"
  }
}

Restoration Event:

{
  "source": "bedrock-budgeteer",
  "detail-type": "Automatic User Restoration Required",
  "detail": {
    "principal_id": "string",
    "restoration_reason": "budget_refresh_period_reached",
    "refresh_date": "2024-01-01T00:00:00Z",
    "current_time": "2024-01-01T00:00:00Z"
  }
}

Configuration Reference

SSM Parameter Reference

Global Parameters:

/bedrock-budgeteer/global/thresholds_percent_warn = "70"
/bedrock-budgeteer/global/thresholds_percent_critical = "90"
/bedrock-budgeteer/global/default_user_budget_usd = "1"
/bedrock-budgeteer/global/grace_period_seconds = "300"

Environment-Specific Parameters:

/bedrock-budgeteer/production/cost/budget_refresh_period_days = "30"

Metric Dimensions

Budget Metrics:

{
  "Environment": "production",
  "PrincipalId": "user-id"  # optional
}

System Metrics:

{
  "Environment": "production",
  "FunctionName": "lambda-function-name"  # optional
}

Extension Points

Adding Custom Lambda Functions 

# In CoreProcessingConstruct
def _create_custom_lambda(self, common_config: Dict[str, Any]) -> None:
    """Create custom Lambda function"""
    
    self.lambda_functions["custom_function"] = lambda_.Function(
        self,
        "CustomFunction",
        function_name=f"bedrock-budgeteer-custom-{self.environment_name}",
        code=lambda_.Code.from_inline(custom_function_code),
        handler="index.lambda_handler",
        **common_config
    )

Adding Custom Metrics 

# In shared utilities
MetricsPublisher.publish_budget_metric(
    "CustomMetric",
    value,
    "Count", 
    {"Environment": "production", "Source": "custom"}
)

Adding Custom Workflow Steps 

# In workflow definition
custom_step = self.create_lambda_invoke_task(
    "CustomStep",
    "custom_function",
    {"input": "data"},
    "$.custom_result"
)

# Chain with other steps
definition = start_step.next(custom_step).next(end_step)

This API reference provides comprehensive documentation for all public interfaces and extension points in the Bedrock Budgeteer system.

Bedrock Budgeteer is an open-source project licensed under MIT.

This site uses Just the Docs, a documentation theme for Jekyll.