Greatest practices for making use of Amazon Bedrock Guardrails to code era workflows

import hashlib
from pathlib import Path

class CodeCommitGuardrail:
    """
    Validates all AI-generated code modifications earlier than they're persevered.
    Analogous to working safety scanners in a Git pre-commit hook.

    Use this when:
    - A coding assistant writes code to disk
    - AI-generated modifications are staged for commit
    - Generated code is about to be deployed
    """

    def __init__(self, consumer, guardrail_id, guardrail_version):
        self.consumer = consumer
        self.guardrail_id = guardrail_id
        self.guardrail_version = guardrail_version
        self.validated_hashes = {}  # Cache: do not re-validate unchanged information

    def validate_file_changes(self, staged_files: dict) -> dict:
        """
        Consider all staged AI-generated code modifications earlier than commit.

        Args:
            staged_files: Dict of {file_path: file_content} for all modified information

        Returns:
            Dict with 'secure' boolean and any violations discovered
        """
        violations = []
        skipped = []

        for file_path, content material in staged_files.objects():
            # Skip information that have not modified since final validation
            content_hash = hashlib.sha256(content material.encode()).hexdigest()
            if self.validated_hashes.get(file_path) == content_hash:
                skipped.append(file_path)
                proceed

            # Consider in 1,000-char aligned chunks
            file_violations = self._evaluate_file(file_path, content material)

            if file_violations:
                violations.prolong(file_violations)
            else:
                # Cache profitable validation
                self.validated_hashes[file_path] = content_hash

        return {
            'secure': len(violations) == 0,
            'violations': violations,
            'files_evaluated': len(staged_files) - len(skipped),
            'files_skipped_cached': len(skipped)
        }

    def _evaluate_file(self, file_path: str, content material: str) -> checklist:
        """Consider a single file's content material towards guardrails."""
        violations = []

        for offset in vary(0, len(content material), 1000):
            chunk = content material[offset:offset + 1000]

            response = self.consumer.apply_guardrail(
                guardrailIdentifier=self.guardrail_id,
                guardrailVersion=self.guardrail_version,
                supply="OUTPUT",
                content material=[{'text': {'text': chunk}}]
            )

            if response['action'] == 'GUARDRAIL_INTERVENED':
                violations.append({
                    'file': file_path,
                    'offset': offset,
                    'size': len(chunk),
                    'assessments': response['assessments'],
                    'snippet': chunk[:200] + '...' if len(chunk) > 200 else chunk
                })

        return violations


# Instance: Integration with a coding assistant's file-write operation
def write_ai_generated_code(file_path: str, content material: str, guardrail: CodeCommitGuardrail):
    """
    Wrapper for file writes that validates content material earlier than persisting.
    """
    outcome = guardrail.validate_file_changes({file_path: content material})

    if not outcome['safe']:
        print(f"Guardrail blocked write to {file_path}")
        for v in outcome['violations']:
            print(f"  Violation at offset {v['offset']}: {v['assessments']}")
        return False

    # Secure to jot down
    Path(file_path).write_text(content material)
    print(f"{file_path} validated and written efficiently")
    return True

# When utilizing InvokeModelWithResponseStream with guardrails
response = bedrock_runtime.invoke_model_with_response_stream(
    modelId='anthropic.claude-sonnet-4-20250514',
    physique=json.dumps(request_body),
    guardrailIdentifier="your-guardrail-id",
    guardrailVersion='1',
    # Key configuration: improve streaming interval
    streamingConfigurations={
        'guardrailStreamingInterval': 1000  # Default is 50!
    }
)

import boto3
import json

bedrock_runtime = boto3.consumer('bedrock-runtime')

def process_coding_request(user_input: str, system_prompt: str, conversation_history: checklist):
    """
    Validate consumer enter with guardrails, then invoke mannequin with out inline scanning.
    The system immediate and dialog historical past are NOT re-evaluated each flip.
    """

    # Step 1: Consider ONLY the brand new dynamic consumer enter
    guardrail_response = bedrock_runtime.apply_guardrail(
        guardrailIdentifier="your-guardrail-id",
        guardrailVersion='1',
        supply="INPUT",
        content material=[
            {
                'text': {
                    'text': user_input  # Only the new content - not the full prompt
                }
            }
        ]
    )

    if guardrail_response['action'] == 'GUARDRAIL_INTERVENED':
        return {
            'blocked': True,
            'message': guardrail_response['outputs'][0]['text'],
            'assessments': guardrail_response['assessments']
        }

    # Step 2: Secure to proceed - invoke mannequin WITHOUT inline guardrails
    # The system immediate and historical past bypass redundant analysis
    messages = conversation_history + [{'role': 'user', 'content': user_input}]

    response = bedrock_runtime.converse(
        modelId='anthropic.claude-sonnet-4-20250514',
        messages=messages,
        system=[{'text': system_prompt}]
        # Be aware: No guardrailConfig right here - we have already validated enter
    )

    return {
        'blocked': False,
        'response': response['output']['message']['content'][0]['text']
    }

def generate_and_validate_code(user_input: str, system_prompt: str):
    """
    Generate code with out inline guardrails, then validate the entire output.
    Very best for detecting secrets and techniques, PII, or unsafe patterns in generated code.
    """

    # Step 1: Generate code with out inline guardrail overhead
    response = bedrock_runtime.converse(
        modelId='anthropic.claude-sonnet-4-20250514',
        messages=[{'role': 'user', 'content': user_input}],
        system=[{'text': system_prompt}]
    )

    generated_code = response['output']['message']['content'][0]['text']

    # Step 2: Validate the COMPLETE code artifact - not intermediate chunks
    guardrail_response = bedrock_runtime.apply_guardrail(
        guardrailIdentifier="your-guardrail-id",
        guardrailVersion='1',
        supply="OUTPUT",
        content material=[
            {
                'text': {
                    'text': generated_code
                }
            }
        ]
    )

    if guardrail_response['action'] == 'GUARDRAIL_INTERVENED':
        return {
            'blocked': True,
            'message': 'Generated code accommodates delicate content material that was blocked.',
            'assessments': guardrail_response['assessments']
        }

    return {
        'blocked': False,
        'code': generated_code
    }

def secure_code_generation_pipeline(user_input: str, system_prompt: str, historical past: checklist):
    """
    Full bidirectional validation with out inline scanning overhead.
    Enter: Test for immediate injection/manipulation
    Output: Test for leaked secrets and techniques and unsafe patterns
    """

    # Step 1: Validate enter for immediate injection
    input_check = bedrock_runtime.apply_guardrail(
        guardrailIdentifier="your-guardrail-id",
        guardrailVersion='1',
        supply="INPUT",
        content material=[{'text': {'text': user_input}}]
    )

    if input_check['action'] == 'GUARDRAIL_INTERVENED':
        return {'blocked': True, 'stage': 'enter', 'purpose': input_check['outputs'][0]['text']}

    # Step 2: Generate code - no inline guardrails wanted
    messages = historical past + [{'role': 'user', 'content': user_input}]
    response = bedrock_runtime.converse(
        modelId='anthropic.claude-sonnet-4-20250514',
        messages=messages,
        system=[{'text': system_prompt}]
    )
    generated_code = response['output']['message']['content'][0]['text']

    # Step 3: Validate output for delicate info
    output_check = bedrock_runtime.apply_guardrail(
        guardrailIdentifier="your-guardrail-id",
        guardrailVersion='1',
        supply="OUTPUT",
        content material=[{'text': {'text': generated_code}}]
    )

    if output_check['action'] == 'GUARDRAIL_INTERVENED':
        return {'blocked': True, 'stage': 'output', 'purpose': output_check['outputs'][0]['text']}

    return {'blocked': False, 'code': generated_code}

import asyncio

class GuardrailBatcher:
    """Accumulates streaming output and evaluates at 1,000-char boundaries."""

    def __init__(self, consumer, guardrail_id, guardrail_version):
        self.consumer = consumer
        self.guardrail_id = guardrail_id
        self.guardrail_version = guardrail_version
        self.buffer = ""
        self.BATCH_SIZE = 1000  # Align to TextUnit boundary

    async def process_chunk(self, chunk: str):
        """Buffer chunks and consider after we hit a TextUnit boundary."""
        self.buffer += chunk

        whereas len(self.buffer) >= self.BATCH_SIZE:
            # Extract precisely 1,000 characters for analysis
            batch = self.buffer[:self.BATCH_SIZE]
            self.buffer = self.buffer[self.BATCH_SIZE:]

            response = self.consumer.apply_guardrail(
                guardrailIdentifier=self.guardrail_id,
                guardrailVersion=self.guardrail_version,
                supply="OUTPUT",
                content material=[{'text': {'text': batch}}]
            )

            if response['action'] == 'GUARDRAIL_INTERVENED':
                elevate GuardrailViolation(response)

    async def flush(self):
        """Consider any remaining content material at finish of stream."""
        if self.buffer:
            response = self.consumer.apply_guardrail(
                guardrailIdentifier=self.guardrail_id,
                guardrailVersion=self.guardrail_version,
                supply="OUTPUT",
                content material=[{'text': {'text': self.buffer}}]
            )
            self.buffer = ""
            return response

import re

class RiskBasedGuardrail:
    """
    Adjusts guardrail analysis technique based mostly on the chance profile
    of the code being generated.

    Excessive-risk code: Full analysis with all safeguards
    Normal code: Light-weight analysis (delicate information solely)
    Low-risk code: Skip intermediate analysis; validate at commit solely
    """

    # Patterns that point out high-risk code
    HIGH_RISK_PATTERNS = [
        r'iam[_.]',                           # IAM coverage manipulation
        r'(access_key|secret_key|password)',   # Credential dealing with
        r'(exec|eval|subprocess|os.system)',  # Code execution
        r'(BEGIN.*PRIVATE KEY)',               # Cryptographic keys
        r'(security_group|nacl|firewall)',     # Community safety
        r'(grant|revoke|permission)',          # Authorization
        r'(encrypt|decrypt|kms)',              # Encryption operations
    ]

    # Patterns that point out low-risk code
    LOW_RISK_PATTERNS = [
        r'(.css|style|className|fontSize)',   # Styling
        r'(render|component|jsx|tsx)',         # UI components
        r'(test|spec|mock|fixture)',           # Test files
        r'(README|docs|comment)',              # Documentation
    ]

    def __init__(self, consumer, guardrail_id_full, guardrail_id_lightweight):
        self.consumer = consumer
        self.guardrail_id_full = guardrail_id_full                # All safeguards lively
        self.guardrail_id_lightweight = guardrail_id_lightweight  # Delicate information solely

    def classify_risk(self, code: str, file_path: str = "") -> str:
        """Decide danger degree of generated code."""
        mixed = code + " " + file_path

        for sample in self.HIGH_RISK_PATTERNS:
            if re.search(sample, mixed, re.IGNORECASE):
                return 'HIGH'

        for sample in self.LOW_RISK_PATTERNS:
            if re.search(sample, mixed, re.IGNORECASE):
                return 'LOW'

        return 'STANDARD'

    def consider(self, code: str, file_path: str = "") -> dict:
        """
        Consider code with acceptable depth based mostly on danger classification.
        """
        risk_level = self.classify_risk(code, file_path)

        if risk_level == 'LOW':
            # Low-risk: defer to commit-time validation
            return {
                'motion': 'PASS',
                'risk_level': 'LOW',
                'analysis': 'deferred_to_commit',
                'purpose': 'Content material categorized as low-risk (UI/checks/docs)'
            }

        # Select guardrail based mostly on danger
        guardrail_id = (
            self.guardrail_id_full if risk_level == 'HIGH'
            else self.guardrail_id_lightweight
        )

        response = self.consumer.apply_guardrail(
            guardrailIdentifier=guardrail_id,
            guardrailVersion='1',
            supply="OUTPUT",
            content material=[{'text': {'text': code}}]
        )

        return {
            'motion': response['action'],
            'risk_level': risk_level,
            'analysis': 'full' if risk_level == 'HIGH' else 'light-weight',
            'assessments': response.get('assessments', [])
        }


# Instance: Utilizing risk-based analysis in a coding pipeline
risk_guardrail = RiskBasedGuardrail(
    consumer=bedrock_runtime,
    guardrail_id_full="guardrail-all-safeguards",
    guardrail_id_lightweight="guardrail-sensitive-info-only"
)

# Excessive-risk: IAM coverage code → full analysis (3 safeguards)
# WARNING: The next is an instance of an INSECURE anti-pattern used right here
# solely to show guardrail detection. By no means use wildcard permissions
# (Motion: '*', Useful resource: '*') in manufacturing. At all times observe the precept of
# least privilege. See: https://docs.aws.amazon.com/IAM/newest/UserGuide/best-practices.html#grant-least-privilege
iam_code = """
coverage = iam.create_policy(
    PolicyName="AdminAccess",
    PolicyDocument=json.dumps({
        'Assertion': [{'Effect': 'Allow', 'Action': '*', 'Resource': '*'}]
    })
)
"""
outcome = risk_guardrail.consider(iam_code, "deploy/iam_policies.py")
# → risk_level: HIGH, analysis: full

# Low-risk: React part → deferred to commit
ui_code = """
perform Button({ label, onClick }) {
    return <button className="btn-primary" onClick={onClick}>{label}</button>;
}
"""
outcome = risk_guardrail.consider(ui_code, "src/elements/Button.tsx")
# → risk_level: LOW, analysis: deferred_to_commit

class AgentPipelineGuardrail:
    """
    Guardrail structure for multi-step agentic coding workflows.

    Key perception: In an agent loop, the mannequin might take 5-10 reasoning steps
    (device calls, observations, reflections) earlier than producing last code.
    Evaluating each intermediate step is wasteful - most comprise inner
    reasoning that by no means reaches the consumer or a file system.

    Technique:
    - INPUT: Validate consumer request earlier than agent begins
    - INTERMEDIATE: Skip chain-of-thought and tool-use reasoning
    - TOOL INPUTS: Validate when agent writes to file or executes code
    - FINAL OUTPUT: Validate the entire response proven to the consumer
    """

    def __init__(self, consumer, guardrail_id, guardrail_version):
        self.consumer = consumer
        self.guardrail_id = guardrail_id
        self.guardrail_version = guardrail_version

    def run_agent_with_guardrails(self, user_request: str, agent):
        """
        Execute an agentic coding workflow with strategic guardrail placement.
        """

        # Gate 1: Validate consumer enter BEFORE agent begins working
        input_check = self._evaluate(user_request, 'INPUT')
        if input_check['action'] == 'GUARDRAIL_INTERVENED':
            return {'blocked': True, 'stage': 'enter', 'response': input_check}

        # Run the agent - NO guardrail analysis throughout reasoning steps
        agent_steps = []
        final_output = None

        for step in agent.run(user_request):
            if step['type'] == 'considering':
                # Skip: inner reasoning does not want analysis
                agent_steps.append(step)
                proceed

            elif step['type'] == 'tool_call':
                # Gate 2: Validate ONLY harmful device inputs
                if self._is_dangerous_tool(step['tool_name']):
                    tool_check = self._evaluate(
                        json.dumps(step['tool_input']), 'OUTPUT'
                    )
                    if tool_check['action'] == 'GUARDRAIL_INTERVENED':
                        return {
                            'blocked': True,
                            'stage': f"tool_call:{step['tool_name']}",
                            'response': tool_check
                        }
                agent_steps.append(step)

            elif step['type'] == 'final_response':
                final_output = step['content']

        # Gate 3: Validate last output earlier than displaying to consumer
        if final_output:
            output_check = self._evaluate(final_output, 'OUTPUT')
            if output_check['action'] == 'GUARDRAIL_INTERVENED':
                return {'blocked': True, 'stage': 'output', 'response': output_check}

        return {
            'blocked': False,
            'output': final_output,
            'steps_taken': len(agent_steps),
            'steps_evaluated': sum(
                1 for s in agent_steps
                if s['type'] == 'tool_call' and self._is_dangerous_tool(s.get('tool_name', ''))
            )
        }

    def _is_dangerous_tool(self, tool_name: str) -> bool:
        """Instruments that write to disk or execute code want guardrail analysis."""
        dangerous_tools = {
            'write_file', 'execute_code', 'run_command',
            'create_file', 'edit_file', 'bash'
        }
        return tool_name in dangerous_tools

    def _evaluate(self, content material: str, supply: str) -> dict:
        return self.consumer.apply_guardrail(
            guardrailIdentifier=self.guardrail_id,
            guardrailVersion=self.guardrail_version,
            supply=supply,
            content material=[{'text': {'text': content}}]
        )