用 Amazon Bedrock 构建可审计、受策略约束的灾难恢复智能体

{
  "service": "payments",
  "source_region": "us-east-1",
  "target_region": "us-west-2",
  "change_ticket": "CHG-4821",
  "requested_action": "failover"
}

python -m venv .venv
source .venv/bin/activate
pip install boto3
export AWS_REGION=us-east-1
export BEDROCK_MODEL_ID='your-enabled-model-id'
python dr_planner.py 'Fail over payments from us-east-1 to us-west-2 under CHG-4821'

import json
import os
import sys
from datetime import datetime, timezone

import boto3

ALLOWED_SERVICES = {"payments", "billing", "identity"}
ALLOWED_TARGETS = {
    "payments": {"us-west-2"},
    "billing": {"us-east-2"},
    "identity": {"us-west-2"},
}

SYSTEM_PROMPT = """You convert disaster-recovery requests into JSON plans.
Return JSON only, with exactly these string fields:
service, source_region, target_region, change_ticket, requested_action.
requested_action must be failover. Never invent missing values; use an empty string.
Do not emit commands, scripts, credentials, or explanatory text.
"""


def build_plan(request_text: str) -> dict:
    client = boto3.client(
        "bedrock-runtime",
        region_name=os.environ.get("AWS_REGION", "us-east-1"),
    )
    response = client.converse(
        modelId=os.environ["BEDROCK_MODEL_ID"],
        system=[{"text": SYSTEM_PROMPT}],
        messages=[{"role": "user", "content": [{"text": request_text}]}],
        inferenceConfig={"temperature": 0, "maxTokens": 300},
    )
    text = response["output"]["message"]["content"][0]["text"]
    return json.loads(text)


def evaluate_policy(plan: dict) -> list[str]:
    required = {
        "service",
        "source_region",
        "target_region",
        "change_ticket",
        "requested_action",
    }
    errors = []
    if set(plan) != required:
        errors.append("plan schema does not match the approved schema")
    if plan.get("service") not in ALLOWED_SERVICES:
        errors.append("service is not approved for automated failover")
    if plan.get("target_region") not in ALLOWED_TARGETS.get(plan.get("service"), set()):
        errors.append("target region is not approved for this service")
    if plan.get("requested_action") != "failover":
        errors.append("unsupported action")
    if not plan.get("change_ticket", "").startswith("CHG-"):
        errors.append("a valid change ticket is required")
    if plan.get("source_region") == plan.get("target_region"):
        errors.append("source and target regions must differ")
    return errors


def main() -> None:
    if len(sys.argv) != 2:
        raise SystemExit("usage: python dr_planner.py '<request>'")

    request_text = sys.argv[1]
    plan = build_plan(request_text)
    errors = evaluate_policy(plan)
    audit_event = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "request": request_text,
        "plan": plan,
        "decision": "REJECTED" if errors else "AWAITING_APPROVAL",
        "policy_errors": errors,
    }
    print(json.dumps(audit_event, indent=2))


if __name__ == "__main__":
    main()