Enterprise AI Service with Guardrails
Build a typed Python service with Bedrock Guardrails, PII masking, and cost observability.
Introduction
30 Second Summary
Every company wants to use AI, but giving employees unrestricted access to a language model is a recipe for leaked secrets, inappropriate content, and runaway costs. Enterprise AI needs guardrails.
In this project, you will build a Python service that connects to Amazon Bedrock, creates safety guardrails programmatically, and tracks every AI interaction for cost and compliance.
What You'll Build
A typed Python connector service that integrates with Amazon Bedrock's Converse API, enforces content filters and PII masking through programmatic guardrails, and logs full observability metrics for every request.
By the end of this project, you'll have:
- A guardrail configuration module that creates content filters, denied topics, and PII masking via the Bedrock API.
- A typed connector service using Pydantic models that wraps the Bedrock Converse API with prompt template management.
- An observability layer that tracks token usage, latency, cost per request, and guardrail interventions.
- Secret Mission: A custom PII regex pattern for employee IDs and automatic cost alerting when usage exceeds a configurable threshold.
Are there any prerequisites?
This project assumes you have Python 3.11 installed (via uv or similar), VS Code ready, and basic fluency with Python (including Pydantic models).
You will also need an AWS account. A Paid plan is recommended for full Bedrock access, and new accounts receive up to $200 in credits.
Set Up AWS and Dev Environment
Before building anything, you need to confirm that your local environment can talk to Amazon Bedrock.
In this step you will verify your AWS CLI credentials, confirm your Python packages are importable, and check that the Amazon Nova Micro model is accessible in your account.
In this step, get ready to:
- Verify your AWS CLI can authenticate with your IAM user.
- Confirm boto3 and pydantic are installed and importable.
- Check that Amazon Nova Micro is enabled in the Bedrock console.
Verify your AWS CLI credentials
Your AWS CLI is already installed and configured with your access key, secret key, and region. The fastest way to confirm everything is wired up correctly is to ask AWS who you are.
- Confirm the CLI is installed by running this command:
aws --version- You should see a version string starting with aws-cli/2 followed by Python and OS info.
- Verify your credentials are working by running this command:
aws sts get-caller-identityYou should see a JSON response containing your Account ID, IAM user ARN, and UserId.
What does this command do?
The aws sts get-caller-identity command asks AWS to confirm which IAM identity is making the request.
If you see a JSON response with your account number and ARN, your access key, secret key, and region are all configured correctly.
Seeing an error about credentials?
If you see a message about being unable to locate credentials, run aws configure again and re-enter your access key, secret key, and set the default region to us-east-1.
Make sure you copied the full access key and secret key from the IAM console with no extra spaces.
Help me troubleshoot my AWS CLI credentials
Verify Python dependencies
Your project uses boto3 to interact with AWS services and Pydantic for typed data models. Both packages are already installed. Let's confirm they import correctly.
- Verify boto3 is importable by running this command:
python -c "import boto3; print(boto3.__version__)"- You should see a version number printed (e.g. 1.43.37 or similar).
- Verify pydantic is importable by running this command:
python -c "import pydantic; print(pydantic.__version__)"You should see a version number printed, confirming pydantic is ready.
Getting a ModuleNotFoundError?
If Python cannot find boto3 or pydantic, reinstall them by running pip install boto3 pydantic.
If you have multiple Python versions installed, make sure you are using the same python binary that pip installed to. Try python3 instead of python if needed.
Help me fix my Python import errors
Confirm Bedrock model access
Amazon Bedrock requires you to explicitly enable each model before you can call it. You have already enabled Amazon Nova Micro in the console. Now let's confirm your CLI can see it.
- Verify Amazon Nova Micro is accessible by running this command:
aws bedrock list-foundation-models --region us-east-1 --query "modelSummaries[?modelId=='amazon.nova-micro-v1:0'].modelId"You should see output containing amazon.nova-micro-v1:0 in the response array.
Why Amazon Nova Micro?
Amazon Nova Micro costs $0.035 per million input tokens and $0.14 per million output tokens. That is roughly 50x cheaper than larger models like Claude.
For a project about guardrails and architecture patterns, the model quality is secondary. You learn the same integration patterns at a fraction of the cost.
Getting an empty array or access denied?
If the response is an empty array [], you need to enable the model. In the AWS console, navigate to Amazon Bedrock then Model access then Manage model access.
Find Amazon Nova Micro in the list and enable it. Model access can take a few minutes to propagate.
Make sure your region is set to us-east-1 in both the console and your CLI configuration.
Help me enable model access in Bedrock
Your environment is ready to go. Next up, you will create enterprise guardrails that filter harmful content, block denied topics, and mask sensitive information.
Create Enterprise Guardrails Programmatically
Your AWS environment is configured and model access is confirmed. Now it's time to define the safety policies that will protect your AI service from generating harmful or sensitive content.
Amazon Bedrock Guardrails let you define safety rules as code instead of clicking through the console. You'll create content filters, a denied topic, and PII masking rules programmatically using the Bedrock management API. This approach gives you version control and makes your guardrails reproducible across environments.
In this step, get ready to:
- Configure content filters that block harmful categories at HIGH strength.
- Define a denied topic and PII masking rules for enterprise compliance.
- Test the guardrail independently without invoking a model.
Build the guardrail creation module
Enterprise AI services need multiple layers of protection. Content filters catch harmful language, denied topics prevent information leakage, and PII masking protects personal data. You'll configure all three in a single API call.
- In VS Code, create a new file called guardrails.py in your project folder.
- Add the following code to define the get_or_create_guardrail function:
import boto3
import json
def get_or_create_guardrail(
region: str = "us-east-1",
guardrail_name: str = "EnterpriseAIGuardrail",
) -> tuple[str, str]:
"""Create or retrieve an enterprise guardrail. Returns (guardrail_id, version)."""
bedrock = boto3.client("bedrock", region_name=region)
# Check if the guardrail already exists (makes the function idempotent)
existing = bedrock.list_guardrails()
for g in existing.get("guardrails", []):
if g["name"] == guardrail_name:
guardrail_id = g["id"]
version = g["version"]
print(f"Found existing guardrail: {guardrail_id} v{version}")
return guardrail_id, version
# Create a new guardrail with content filters, denied topics, and PII masking
response = bedrock.create_guardrail(
name=guardrail_name,
description="Enterprise guardrail with content filters, denied topics, and PII masking",
blockedInputMessaging="Your request was blocked by our content policy. Please rephrase your message.",
blockedOutputsMessaging="The model response was blocked by our content policy.",
contentPolicyConfig={
"filtersConfig": [
{"type": "SEXUAL", "inputStrength": "HIGH", "outputStrength": "HIGH", "inputAction": "BLOCK", "outputAction": "BLOCK", "inputModalities": ["TEXT"], "outputModalities": ["TEXT"]},
{"type": "VIOLENCE", "inputStrength": "HIGH", "outputStrength": "HIGH", "inputAction": "BLOCK", "outputAction": "BLOCK", "inputModalities": ["TEXT"], "outputModalities": ["TEXT"]},
{"type": "HATE", "inputStrength": "HIGH", "outputStrength": "HIGH", "inputAction": "BLOCK", "outputAction": "BLOCK", "inputModalities": ["TEXT"], "outputModalities": ["TEXT"]},
{"type": "INSULTS", "inputStrength": "HIGH", "outputStrength": "HIGH", "inputAction": "BLOCK", "outputAction": "BLOCK", "inputModalities": ["TEXT"], "outputModalities": ["TEXT"]},
{"type": "MISCONDUCT", "inputStrength": "HIGH", "outputStrength": "HIGH", "inputAction": "BLOCK", "outputAction": "BLOCK", "inputModalities": ["TEXT"], "outputModalities": ["TEXT"]},
]
},
topicPolicyConfig={
"topicsConfig": [
{
"name": "CompetitorDiscussion",
"definition": "Any discussion about competitor products, services, or companies that could reveal strategic information or bias recommendations.",
"examples": ["What do you think about Google Cloud vs AWS?", "Compare our product to the competitor's offering.", "Should we switch to a competitor's solution?"],
"type": "DENY",
}
]
},
sensitiveInformationPolicyConfig={
"piiEntitiesConfig": [
{"type": "EMAIL", "action": "ANONYMIZE"},
{"type": "PHONE", "action": "ANONYMIZE"},
{"type": "US_SOCIAL_SECURITY_NUMBER", "action": "BLOCK"},
]
},
)
guardrail_id = response["guardrailId"]
print(f"Created guardrail: {guardrail_id}")
# Create a versioned snapshot so the guardrail can be applied at runtime
version_response = bedrock.create_guardrail_version(
guardrailIdentifier=guardrail_id,
description="Initial version with content filters, denied topics, and PII masking",
)
version = version_response["version"]
print(f"Created guardrail version: {version}")
return guardrail_id, versionWhat does this code do?
This function handles three responsibilities in one call to create_guardrail:
- Content filters block five harmful categories (SEXUAL, VIOLENCE, HATE, INSULTS, MISCONDUCT) at HIGH strength for both input and output.
- The denied topic CompetitorDiscussion prevents the model from engaging in competitor comparisons that could leak strategic information.
- PII masking anonymizes email addresses and phone numbers, and fully blocks Social Security numbers.
- The function first checks list_guardrails for an existing guardrail with the same name, making it safe to run multiple times without creating duplicates.
- After creation, create_guardrail_version takes a snapshot so the guardrail can be referenced at a specific version during inference.
Why create guardrails as code?
Defining guardrails programmatically means you can store them in version control, reproduce them across AWS accounts, and automate updates. Clicking through the Bedrock console works for experimentation, but production AI services need repeatable, auditable configuration.
Add the standalone test function
Before wiring guardrails into your full connector, you can test them independently using the ApplyGuardrail API. This evaluates text against your guardrail policies without invoking a model, so you pay only for the guardrail evaluation cost.
- In guardrails.py, add the following function below the get_or_create_guardrail function:
def test_guardrail_standalone(
guardrail_id: str, guardrail_version: str, text: str, region: str = "us-east-1"
) -> dict:
"""Test a guardrail without invoking a model using ApplyGuardrail."""
bedrock_runtime = boto3.client("bedrock-runtime", region_name=region)
# Evaluate the text against the guardrail without calling a model
response = bedrock_runtime.apply_guardrail(
guardrailIdentifier=guardrail_id,
guardrailVersion=guardrail_version,
source="INPUT",
content=[{"text": {"text": text}}],
)
print(f"Guardrail action: {response['action']}")
if response.get("assessments"):
print(f"Assessments: {json.dumps(response['assessments'], indent=2)}")
return responseWhat does this function do?
The apply_guardrail method sends text to your guardrail for evaluation without invoking any foundation model. It uses the bedrock-runtime client (not the management client used for creation).
The response includes an action field that is either GUARDRAIL_INTERVENED (content was blocked or masked) or NONE (content passed all checks). The assessments array shows exactly which policy triggered.
- Save guardrails.py.
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before running the test.
ⓧ I'd like to double check the full code
Here is the complete guardrails.py file:
import boto3
import json
def get_or_create_guardrail(
region: str = "us-east-1",
guardrail_name: str = "EnterpriseAIGuardrail",
) -> tuple[str, str]:
"""Create or retrieve an enterprise guardrail. Returns (guardrail_id, version)."""
bedrock = boto3.client("bedrock", region_name=region)
existing = bedrock.list_guardrails()
for g in existing.get("guardrails", []):
if g["name"] == guardrail_name:
guardrail_id = g["id"]
version = g["version"]
print(f"Found existing guardrail: {guardrail_id} v{version}")
return guardrail_id, version
response = bedrock.create_guardrail(
name=guardrail_name,
description="Enterprise guardrail with content filters, denied topics, and PII masking",
blockedInputMessaging="Your request was blocked by our content policy. Please rephrase your message.",
blockedOutputsMessaging="The model response was blocked by our content policy.",
contentPolicyConfig={
"filtersConfig": [
{"type": "SEXUAL", "inputStrength": "HIGH", "outputStrength": "HIGH", "inputAction": "BLOCK", "outputAction": "BLOCK", "inputModalities": ["TEXT"], "outputModalities": ["TEXT"]},
{"type": "VIOLENCE", "inputStrength": "HIGH", "outputStrength": "HIGH", "inputAction": "BLOCK", "outputAction": "BLOCK", "inputModalities": ["TEXT"], "outputModalities": ["TEXT"]},
{"type": "HATE", "inputStrength": "HIGH", "outputStrength": "HIGH", "inputAction": "BLOCK", "outputAction": "BLOCK", "inputModalities": ["TEXT"], "outputModalities": ["TEXT"]},
{"type": "INSULTS", "inputStrength": "HIGH", "outputStrength": "HIGH", "inputAction": "BLOCK", "outputAction": "BLOCK", "inputModalities": ["TEXT"], "outputModalities": ["TEXT"]},
{"type": "MISCONDUCT", "inputStrength": "HIGH", "outputStrength": "HIGH", "inputAction": "BLOCK", "outputAction": "BLOCK", "inputModalities": ["TEXT"], "outputModalities": ["TEXT"]},
]
},
topicPolicyConfig={
"topicsConfig": [
{
"name": "CompetitorDiscussion",
"definition": "Any discussion about competitor products, services, or companies that could reveal strategic information or bias recommendations.",
"examples": ["What do you think about Google Cloud vs AWS?", "Compare our product to the competitor's offering.", "Should we switch to a competitor's solution?"],
"type": "DENY",
}
]
},
sensitiveInformationPolicyConfig={
"piiEntitiesConfig": [
{"type": "EMAIL", "action": "ANONYMIZE"},
{"type": "PHONE", "action": "ANONYMIZE"},
{"type": "US_SOCIAL_SECURITY_NUMBER", "action": "BLOCK"},
]
},
)
guardrail_id = response["guardrailId"]
print(f"Created guardrail: {guardrail_id}")
version_response = bedrock.create_guardrail_version(
guardrailIdentifier=guardrail_id,
description="Initial version with content filters, denied topics, and PII masking",
)
version = version_response["version"]
print(f"Created guardrail version: {version}")
return guardrail_id, version
def test_guardrail_standalone(
guardrail_id: str, guardrail_version: str, text: str, region: str = "us-east-1"
) -> dict:
"""Test a guardrail without invoking a model using ApplyGuardrail."""
bedrock_runtime = boto3.client("bedrock-runtime", region_name=region)
response = bedrock_runtime.apply_guardrail(
guardrailIdentifier=guardrail_id,
guardrailVersion=guardrail_version,
source="INPUT",
content=[{"text": {"text": text}}],
)
print(f"Guardrail action: {response['action']}")
if response.get("assessments"):
print(f"Assessments: {json.dumps(response['assessments'], indent=2)}")
return responseRun your guardrail test
Time to create the guardrail in your AWS account and verify it blocks harmful content. This call will hit the Bedrock API, create the guardrail resource, and then test it with a violent prompt.
- Run the guardrail creation and test by executing this command in your terminal:
python -c "from guardrails import get_or_create_guardrail, test_guardrail_standalone; gid, gv = get_or_create_guardrail(); test_guardrail_standalone(gid, gv, 'I want to harm someone violently.')"- You should see output showing the guardrail was created with an ID and version number, followed by Guardrail action: GUARDRAIL_INTERVENED.
- The assessments JSON shows which content filter triggered (VIOLENCE with HIGH confidence).
What should I see?
Your terminal output will look something like this (the guardrail ID will be different):
Created guardrail: abc123def456
Created guardrail version: 1
Guardrail action: GUARDRAIL_INTERVENED
Assessments: [{"contentPolicy": {"filters": [{"type": "VIOLENCE", ...}]}}]The key confirmation is that action equals GUARDRAIL_INTERVENED. This proves your content filters are working before you connect them to a model.
Seeing an AccessDeniedException?
- Verify your IAM user has the AmazonBedrockFullAccess policy attached. You configured this in Step 1.
- Check that your aws configure region matches us-east-1. The guardrail creation uses the region from the function default, but your CLI credentials must also be valid for that region.
- If you see a throttling error, wait 30 seconds and run the command again. The function is idempotent, so it will find the existing guardrail on the second run.
My guardrail test is failing with an access error
- Run the command a second time to confirm idempotency. You should see Found existing guardrail instead of creating a new one.
Your guardrail is live and blocking harmful content. Next up, you'll build a typed connector service that wraps the Bedrock Converse API and applies this guardrail on every model invocation.
Build the Typed AI Connector Service
Your guardrail is live and tested. Now you need a clean way to call Amazon Bedrock that enforces type safety on every request and response.
In this step, you will build a typed connector service using Pydantic models for validation and a prompt template system for reusable system prompts. The connector wraps the Bedrock Converse API and automatically applies your guardrail to every invocation.
In this step, get ready to:
- Create Pydantic models for typed request and response validation.
- Build the BedrockConnector class that wraps the Converse API with guardrails.
- Implement prompt template management with variable substitution.
Create Pydantic models for request and response validation
Enterprise services need strict contracts between components. Pydantic models define exactly what data flows in and out of your connector, catching invalid inputs before they reach the API.
- Create a new file called models.py in your project folder.
- Add the imports and the first two model classes by pasting this code:
from pydantic import BaseModel, Field, field_validator
from typing import Optional
class GuardrailConfig(BaseModel):
"""Holds the guardrail identifier and version for API calls."""
guardrail_id: str
guardrail_version: str
trace_enabled: bool = True
class PromptTemplate(BaseModel):
"""A versioned system prompt with variable placeholders."""
name: str
version: str
system_prompt: str
required_variables: list[str] = Field(default_factory=list)
def render(self, variables: dict[str, str]) -> str:
# Check that all required variables are provided
missing = [v for v in self.required_variables if v not in variables]
if missing:
raise ValueError(f"Missing required variables: {missing}")
# Replace {{variable}} placeholders with actual values
rendered = self.system_prompt
for key, value in variables.items():
rendered = rendered.replace(f"{{{{{key}}}}}", value)
return renderedWhat does this code do?
GuardrailConfig stores the guardrail ID and version so the connector can attach them to every API call. The trace_enabled flag controls whether Bedrock returns detailed trace data about guardrail decisions.
PromptTemplate defines a reusable system prompt with a name, version, and variable placeholders. The render() method substitutes {{variable}} placeholders with real values, raising an error if any required variable is missing.
- Now add the request and response models below the PromptTemplate class in the same file:
class ConversationRequest(BaseModel):
"""Validated request to the AI connector."""
template_name: Optional[str] = None
template_variables: dict[str, str] = Field(default_factory=dict)
system_prompt: Optional[str] = None
user_message: str
max_tokens: int = Field(default=512, gt=0)
temperature: float = Field(default=0.5, ge=0.0, le=1.0)
@field_validator("user_message")
@classmethod
def message_not_empty(cls, v: str) -> str:
# Reject blank or whitespace-only messages
if not v.strip():
raise ValueError("user_message must not be empty")
return v
class ConversationResponse(BaseModel):
"""Structured response from the AI connector."""
response_text: str
stop_reason: str
input_tokens: int
output_tokens: int
latency_ms: int
guardrail_intervened: bool
cost_usd: float
trace: Optional[dict] = NoneWhat does this code do?
ConversationRequest defines everything needed to make a model call. Pydantic's Field constraints enforce that temperature stays between 0 and 1, and max_tokens is always positive.
The @field_validator decorator rejects empty messages before they reach the API. ConversationResponse captures the model's reply along with token counts, latency, cost, and whether the guardrail intervened.
- Save models.py.
- Verify the models load correctly by running this in your terminal:
python -c "from models import ConversationRequest, GuardrailConfig; print('Models loaded successfully')"You should see Models loaded successfully printed in the terminal.
Seeing a ModuleNotFoundError?
Make sure your terminal is in the same folder as models.py. Run ls to confirm the file exists in your current directory.
If you see a Pydantic import error, confirm you installed it earlier with pip install pydantic.
Help me debug my models.py import error
✔️ Awesome, I've got everything!
Great. Make sure you have saved your file before moving on.
ⓧ I'd like to double check the full code
Here is the complete models.py for comparison:
from pydantic import BaseModel, Field, field_validator
from typing import Optional
class GuardrailConfig(BaseModel):
guardrail_id: str
guardrail_version: str
trace_enabled: bool = True
class PromptTemplate(BaseModel):
name: str
version: str
system_prompt: str
required_variables: list[str] = Field(default_factory=list)
def render(self, variables: dict[str, str]) -> str:
missing = [v for v in self.required_variables if v not in variables]
if missing:
raise ValueError(f"Missing required variables: {missing}")
rendered = self.system_prompt
for key, value in variables.items():
rendered = rendered.replace(f"{{{{{key}}}}}", value)
return rendered
class ConversationRequest(BaseModel):
template_name: Optional[str] = None
template_variables: dict[str, str] = Field(default_factory=dict)
system_prompt: Optional[str] = None
user_message: str
max_tokens: int = Field(default=512, gt=0)
temperature: float = Field(default=0.5, ge=0.0, le=1.0)
@field_validator("user_message")
@classmethod
def message_not_empty(cls, v: str) -> str:
if not v.strip():
raise ValueError("user_message must not be empty")
return v
class ConversationResponse(BaseModel):
response_text: str
stop_reason: str
input_tokens: int
output_tokens: int
latency_ms: int
guardrail_intervened: bool
cost_usd: float
trace: Optional[dict] = NoneBuild the BedrockConnector class
The connector class is the central piece of your service. It holds the Bedrock runtime client, applies your guardrail to every call, and records metrics for observability.
- Create a new file called connector.py in your project folder.
- Add the imports, class declaration, and initializer by pasting this code:
import boto3
from models import (
ConversationRequest,
ConversationResponse,
GuardrailConfig,
PromptTemplate,
)
from observability import MetricsCollector
class BedrockConnector:
MODEL_ID = "amazon.nova-micro-v1:0"
def __init__(
self,
guardrail_config: GuardrailConfig,
metrics_collector: MetricsCollector,
region: str = "us-east-1",
):
# Create the Bedrock runtime client for model invocations
self.client = boto3.client("bedrock-runtime", region_name=region)
self.guardrail_config = guardrail_config
self.metrics = metrics_collector
# Registry to store named prompt templates
self.prompt_registry: dict[str, PromptTemplate] = {}
def register_template(self, template: PromptTemplate) -> None:
"""Register a prompt template for use in conversations."""
self.prompt_registry[template.name] = templateWhat does this code do?
The class stores a Bedrock runtime client (bedrock-runtime) for inference operations, a GuardrailConfig for attaching guardrails, and a MetricsCollector for recording usage after each call.
MODEL_ID is set to amazon.nova-micro-v1:0, the cheapest Bedrock model at $0.035 per million input tokens. The prompt_registry dict will hold named templates you register later.
Why does it import from observability?
The from observability import MetricsCollector line references a module you have not created yet. You will build observability.py in the next step. For now, the import is intentional. You are building the modules in dependency order before wiring them together.
- Add the converse method below register_template in the same class:
def converse(self, request: ConversationRequest) -> ConversationResponse:
"""Send a validated request to Bedrock with guardrails applied."""
system_prompt = self._resolve_system_prompt(request)
# Format the user message for the Converse API
messages = [
{"role": "user", "content": [{"text": request.user_message}]}
]
system_config = [{"text": system_prompt}] if system_prompt else None
# Attach guardrail configuration to the request
guardrail_cfg = {
"guardrailIdentifier": self.guardrail_config.guardrail_id,
"guardrailVersion": self.guardrail_config.guardrail_version,
"trace": "enabled" if self.guardrail_config.trace_enabled else "disabled",
}
# Build the Converse API keyword arguments
kwargs = {
"modelId": self.MODEL_ID,
"messages": messages,
"inferenceConfig": {
"maxTokens": request.max_tokens,
"temperature": request.temperature,
},
"guardrailConfig": guardrail_cfg,
}
if system_config:
kwargs["system"] = system_config
response = self.client.converse(**kwargs)
# Extract response fields from the Converse API response
response_text = response["output"]["message"]["content"][0]["text"]
stop_reason = response["stopReason"]
usage = response["usage"]
latency_ms = response["metrics"]["latencyMs"]
guardrail_intervened = stop_reason == "guardrail_intervened"
trace_data = response.get("trace", {}).get("guardrail")
# Calculate cost and record metrics
cost = self.metrics.calculate_cost(
usage["inputTokens"], usage["outputTokens"]
)
self.metrics.record(
model_id=self.MODEL_ID,
input_tokens=usage["inputTokens"],
output_tokens=usage["outputTokens"],
latency_ms=latency_ms,
guardrail_intervened=guardrail_intervened,
stop_reason=stop_reason,
)
return ConversationResponse(
response_text=response_text,
stop_reason=stop_reason,
input_tokens=usage["inputTokens"],
output_tokens=usage["outputTokens"],
latency_ms=latency_ms,
guardrail_intervened=guardrail_intervened,
cost_usd=cost,
trace=trace_data,
)What does the converse method do?
This method is the core of your connector. It resolves the system prompt (from a template or a raw string), formats the request for the Converse API, and attaches your guardrail via guardrailConfig.
After calling client.converse(), it extracts the response text, token usage, latency, and stop reason. If stopReason equals guardrail_intervened, the guardrail blocked the content. Every call is recorded to the metrics collector for observability.
Why use the Converse API instead of InvokeModel?
AWS recommends the Converse API over InvokeModel because it provides a consistent interface across all Bedrock models. It includes standardized usage metrics in every response and has native guardrailConfig support. This is the enterprise integration pattern.
- Save connector.py.
You will not be able to run this file yet because observability.py does not exist. You will create it in the next step.
Seeing a syntax error in your code?
Check that the converse method is indented one level inside the BedrockConnector class. Python is strict about indentation.
Verify the dictionary keys match exactly. The Converse API expects guardrailIdentifier and guardrailVersion (camelCase), not snake_case.
Help me fix a syntax error in my connector.py file
Implement prompt template resolution
The connector needs a way to look up registered templates by name and fall back to a raw system prompt when no template is specified. This keeps prompt management separate from invocation logic.
- Add these two helper methods below the converse method in connector.py:
def _resolve_system_prompt(self, request: ConversationRequest) -> str | None:
"""Look up a template by name, or fall back to the raw system prompt."""
if request.template_name and request.template_name in self.prompt_registry:
template = self.prompt_registry[request.template_name]
return template.render(request.template_variables)
return request.system_prompt
def get_usage_report(self) -> dict:
"""Return the cumulative usage report from the metrics collector."""
return self.metrics.usage_report()What does this code do?
_resolve_system_prompt checks if the request specifies a template_name. If it finds a matching template in the registry, it renders it with the provided variables. Otherwise, it returns the raw system_prompt string (or None).
get_usage_report delegates to the metrics collector, giving you a single method to retrieve all observability data from the connector.
- Save connector.py.
- Confirm both files exist by running:
lsYou should see models.py, connector.py, and guardrails.py in the output.
Missing one of the files?
Make sure you created the file in the same project folder as guardrails.py. If you accidentally saved it elsewhere, move it into the correct directory.
Help me find where my connector.py file was saved
✔️ Awesome, I've got everything!
Great. Double check that both files are saved before continuing.
ⓧ I'd like to double check the full code
Here is the complete connector.py for comparison:
import boto3
from models import (
ConversationRequest,
ConversationResponse,
GuardrailConfig,
PromptTemplate,
)
from observability import MetricsCollector
class BedrockConnector:
MODEL_ID = "amazon.nova-micro-v1:0"
def __init__(
self,
guardrail_config: GuardrailConfig,
metrics_collector: MetricsCollector,
region: str = "us-east-1",
):
self.client = boto3.client("bedrock-runtime", region_name=region)
self.guardrail_config = guardrail_config
self.metrics = metrics_collector
self.prompt_registry: dict[str, PromptTemplate] = {}
def register_template(self, template: PromptTemplate) -> None:
self.prompt_registry[template.name] = template
def converse(self, request: ConversationRequest) -> ConversationResponse:
system_prompt = self._resolve_system_prompt(request)
messages = [
{"role": "user", "content": [{"text": request.user_message}]}
]
system_config = [{"text": system_prompt}] if system_prompt else None
guardrail_cfg = {
"guardrailIdentifier": self.guardrail_config.guardrail_id,
"guardrailVersion": self.guardrail_config.guardrail_version,
"trace": "enabled" if self.guardrail_config.trace_enabled else "disabled",
}
kwargs = {
"modelId": self.MODEL_ID,
"messages": messages,
"inferenceConfig": {
"maxTokens": request.max_tokens,
"temperature": request.temperature,
},
"guardrailConfig": guardrail_cfg,
}
if system_config:
kwargs["system"] = system_config
response = self.client.converse(**kwargs)
response_text = response["output"]["message"]["content"][0]["text"]
stop_reason = response["stopReason"]
usage = response["usage"]
latency_ms = response["metrics"]["latencyMs"]
guardrail_intervened = stop_reason == "guardrail_intervened"
trace_data = response.get("trace", {}).get("guardrail")
cost = self.metrics.calculate_cost(
usage["inputTokens"], usage["outputTokens"]
)
self.metrics.record(
model_id=self.MODEL_ID,
input_tokens=usage["inputTokens"],
output_tokens=usage["outputTokens"],
latency_ms=latency_ms,
guardrail_intervened=guardrail_intervened,
stop_reason=stop_reason,
)
return ConversationResponse(
response_text=response_text,
stop_reason=stop_reason,
input_tokens=usage["inputTokens"],
output_tokens=usage["outputTokens"],
latency_ms=latency_ms,
guardrail_intervened=guardrail_intervened,
cost_usd=cost,
trace=trace_data,
)
def _resolve_system_prompt(self, request: ConversationRequest) -> str | None:
if request.template_name and request.template_name in self.prompt_registry:
template = self.prompt_registry[request.template_name]
return template.render(request.template_variables)
return request.system_prompt
def get_usage_report(self) -> dict:
return self.metrics.usage_report()Your typed connector is ready to invoke Bedrock with guardrails on every call. Next up, you will build the observability module that tracks tokens, latency, and cost for every request.
Implement Observability and Cost Optimization
Your connector can invoke models and enforce guardrails. But in an enterprise setting, you need to know exactly how much each request costs, how fast it responds, and how often guardrails fire.
In this step, you will build an observability layer that tracks every interaction. It calculates per-request cost using Amazon Nova Micro pricing, monitors latency, and alerts you when cumulative spending approaches a budget threshold.
In this step, get ready to:
- Create an observability module with per-request cost calculation.
- Verify the connector integration that records metrics automatically.
- Implement budget threshold alerting with WARNING and CRITICAL levels.
Create the observability module with metrics tracking
Every AI request has a measurable cost. Amazon Nova Micro charges $0.035 per million input tokens and $0.14 per million output tokens. Your observability module will use these rates to calculate exact costs for every request.
- Create a new file called observability.py in your project folder.
- Add the pricing constants and the RequestMetrics dataclass by pasting this code:
from dataclasses import dataclass, field
from datetime import datetime, timezone
# Amazon Nova Micro pricing per token
NOVA_MICRO_INPUT_PRICE_PER_TOKEN = 0.035 / 1_000_000
NOVA_MICRO_OUTPUT_PRICE_PER_TOKEN = 0.14 / 1_000_000
@dataclass
class RequestMetrics:
timestamp: str
model_id: str
input_tokens: int
output_tokens: int
latency_ms: int
cost_usd: float
guardrail_intervened: bool
stop_reason: strWhat does this code do?
The two pricing constants convert the per-million-token rates into per-token rates. This makes cost calculation a simple multiplication later.
The RequestMetrics dataclass holds everything you need to know about a single AI request: when it happened, how many tokens it used, how long it took, what it cost, and whether the guardrail intervened.
- Below the RequestMetrics class, add the MetricsCollector class that tracks all metrics and enforces budget alerts:
@dataclass
class MetricsCollector:
budget_usd: float = 5.0
alert_thresholds: list[float] = field(default_factory=lambda: [0.5, 0.8, 1.0])
metrics: list[RequestMetrics] = field(default_factory=list)
alerts_fired: list[str] = field(default_factory=list)
def calculate_cost(self, input_tokens: int, output_tokens: int) -> float:
input_cost = input_tokens * NOVA_MICRO_INPUT_PRICE_PER_TOKEN
output_cost = output_tokens * NOVA_MICRO_OUTPUT_PRICE_PER_TOKEN
return round(input_cost + output_cost, 8)
def record(
self,
model_id: str,
input_tokens: int,
output_tokens: int,
latency_ms: int,
guardrail_intervened: bool,
stop_reason: str,
) -> RequestMetrics:
cost = self.calculate_cost(input_tokens, output_tokens)
metric = RequestMetrics(
timestamp=datetime.now(timezone.utc).isoformat(),
model_id=model_id,
input_tokens=input_tokens,
output_tokens=output_tokens,
latency_ms=latency_ms,
cost_usd=cost,
guardrail_intervened=guardrail_intervened,
stop_reason=stop_reason,
)
self.metrics.append(metric)
self._check_budget_alerts()
return metric
def _check_budget_alerts(self) -> None:
total = self.total_cost
for threshold in self.alert_thresholds:
level = "INFO" if threshold <= 0.5 else "WARNING" if threshold <= 0.8 else "CRITICAL"
alert_key = f"{level}_{threshold}"
if total >= self.budget_usd * threshold and alert_key not in self.alerts_fired:
msg = f"[{level}] Cost alert: ${total:.6f} has reached {threshold*100:.0f}% of ${self.budget_usd:.2f} budget"
print(msg)
self.alerts_fired.append(alert_key)
@property
def total_cost(self) -> float:
return sum(m.cost_usd for m in self.metrics)
@property
def total_tokens(self) -> int:
return sum(m.input_tokens + m.output_tokens for m in self.metrics)
@property
def average_latency(self) -> float:
if not self.metrics:
return 0.0
return sum(m.latency_ms for m in self.metrics) / len(self.metrics)
@property
def guardrail_intervention_rate(self) -> float:
if not self.metrics:
return 0.0
interventions = sum(1 for m in self.metrics if m.guardrail_intervened)
return interventions / len(self.metrics)
def usage_report(self) -> dict:
return {
"total_requests": len(self.metrics),
"total_tokens": self.total_tokens,
"total_cost_usd": round(self.total_cost, 6),
"average_latency_ms": round(self.average_latency, 1),
"guardrail_intervention_rate": round(self.guardrail_intervention_rate, 3),
"budget_usd": self.budget_usd,
"budget_remaining_usd": round(self.budget_usd - self.total_cost, 6),
"alerts_fired": self.alerts_fired,
}What does this code do?
The MetricsCollector is the central observability hub. Here is what each part does:
- calculate_cost() multiplies token counts by the per-token price constants to get exact USD cost.
- record() creates a timestamped metric entry and appends it to the list. After recording, it checks budget alerts.
- _check_budget_alerts() compares cumulative cost against thresholds (50%, 80%, 100%) and prints a warning when crossed.
- usage_report() returns a summary dict with totals, averages, and remaining budget for dashboarding.
- Save observability.py.
- Verify the module loads without errors by running:
python -c "from observability import MetricsCollector; print('observability module loaded')"You should see observability module loaded printed with no import errors.
Seeing an import error?
Check that observability.py is saved in the same folder as your other project files. The filename must be exactly observability.py with no typos.
If you see a SyntaxError, double-check that the @dataclass decorator appears directly above each class definition.
Help me debug my observability.py import error
✔️ Awesome, I've got everything!
Great. Make sure you have saved observability.py before moving on.
ⓧ I'd like to double check the full code
Here is the complete observability.py file. Compare your code against this:
from dataclasses import dataclass, field
from datetime import datetime, timezone
NOVA_MICRO_INPUT_PRICE_PER_TOKEN = 0.035 / 1_000_000
NOVA_MICRO_OUTPUT_PRICE_PER_TOKEN = 0.14 / 1_000_000
@dataclass
class RequestMetrics:
timestamp: str
model_id: str
input_tokens: int
output_tokens: int
latency_ms: int
cost_usd: float
guardrail_intervened: bool
stop_reason: str
@dataclass
class MetricsCollector:
budget_usd: float = 5.0
alert_thresholds: list[float] = field(default_factory=lambda: [0.5, 0.8, 1.0])
metrics: list[RequestMetrics] = field(default_factory=list)
alerts_fired: list[str] = field(default_factory=list)
def calculate_cost(self, input_tokens: int, output_tokens: int) -> float:
input_cost = input_tokens * NOVA_MICRO_INPUT_PRICE_PER_TOKEN
output_cost = output_tokens * NOVA_MICRO_OUTPUT_PRICE_PER_TOKEN
return round(input_cost + output_cost, 8)
def record(
self,
model_id: str,
input_tokens: int,
output_tokens: int,
latency_ms: int,
guardrail_intervened: bool,
stop_reason: str,
) -> RequestMetrics:
cost = self.calculate_cost(input_tokens, output_tokens)
metric = RequestMetrics(
timestamp=datetime.now(timezone.utc).isoformat(),
model_id=model_id,
input_tokens=input_tokens,
output_tokens=output_tokens,
latency_ms=latency_ms,
cost_usd=cost,
guardrail_intervened=guardrail_intervened,
stop_reason=stop_reason,
)
self.metrics.append(metric)
self._check_budget_alerts()
return metric
def _check_budget_alerts(self) -> None:
total = self.total_cost
for threshold in self.alert_thresholds:
level = "INFO" if threshold <= 0.5 else "WARNING" if threshold <= 0.8 else "CRITICAL"
alert_key = f"{level}_{threshold}"
if total >= self.budget_usd * threshold and alert_key not in self.alerts_fired:
msg = f"[{level}] Cost alert: ${total:.6f} has reached {threshold*100:.0f}% of ${self.budget_usd:.2f} budget"
print(msg)
self.alerts_fired.append(alert_key)
@property
def total_cost(self) -> float:
return sum(m.cost_usd for m in self.metrics)
@property
def total_tokens(self) -> int:
return sum(m.input_tokens + m.output_tokens for m in self.metrics)
@property
def average_latency(self) -> float:
if not self.metrics:
return 0.0
return sum(m.latency_ms for m in self.metrics) / len(self.metrics)
@property
def guardrail_intervention_rate(self) -> float:
if not self.metrics:
return 0.0
interventions = sum(1 for m in self.metrics if m.guardrail_intervened)
return interventions / len(self.metrics)
def usage_report(self) -> dict:
return {
"total_requests": len(self.metrics),
"total_tokens": self.total_tokens,
"total_cost_usd": round(self.total_cost, 6),
"average_latency_ms": round(self.average_latency, 1),
"guardrail_intervention_rate": round(self.guardrail_intervention_rate, 3),
"budget_usd": self.budget_usd,
"budget_remaining_usd": round(self.budget_usd - self.total_cost, 6),
"alerts_fired": self.alerts_fired,
}Verify the connector integration
Your BedrockConnector in connector.py already imports and uses MetricsCollector. Every time converse() runs, it automatically calculates cost and records metrics.
- Switch back to connector.py and find this section inside the converse method:
cost = self.metrics.calculate_cost(
usage["inputTokens"], usage["outputTokens"]
)
self.metrics.record(
model_id=self.MODEL_ID,
input_tokens=usage["inputTokens"],
output_tokens=usage["outputTokens"],
latency_ms=latency_ms,
guardrail_intervened=guardrail_intervened,
stop_reason=stop_reason,
)How does the integration work?
After every Converse API call returns, the connector extracts token counts and latency from the response. It passes these values to your MetricsCollector which calculates cost and stores the metric.
The get_usage_report() method on the connector delegates to self.metrics.usage_report(). This gives you a single method call to get a full summary of all requests made through that connector instance.
- Confirm that connector.py already imports MetricsCollector at the top of the file:
from observability import MetricsCollectorNo changes to connector.py are needed. The integration was built into the connector in the previous step.
- Verify both modules work together by running:
python -c "from connector import BedrockConnector; print('connector + observability loaded')"You should see connector + observability loaded with no import errors.
Seeing a ModuleNotFoundError?
If the error mentions observability, make sure you saved observability.py in the same directory as connector.py. Python looks for modules in the current directory first.
If the error mentions pydantic or boto3, verify your packages are installed by running pip install boto3 pydantic.
Help me fix my import error when loading connector.py
Test the cost budget alerting
The _check_budget_alerts() method fires at three thresholds: INFO at 50%, WARNING at 80%, and CRITICAL at 100% of the configured budget. Each alert fires only once, preventing duplicate warnings.
- Verify the alerting works by running this quick test:
python -c "
from observability import MetricsCollector
m = MetricsCollector(budget_usd=0.001)
m.record('test', 5000, 5000, 100, False, 'end_turn')
print('Alerts:', m.alerts_fired)
"You should see one or more alert messages printed, followed by the alerts list showing which thresholds were crossed.
Why a tiny budget for this test?
By setting budget_usd=0.001 (one tenth of a cent), even a small number of tokens exceeds all three thresholds immediately. This proves the alerting logic works without spending real money on API calls.
In production, you would set a realistic budget like $5.00. The system would then warn you as costs accumulate across many requests.
Not seeing any alerts?
Check that your MetricsCollector class has the alert_thresholds field with default values [0.5, 0.8, 1.0]. Also verify that record() calls self._check_budget_alerts() at the end.
Help me debug why my budget alerts are not firing
Your observability layer is locked in. Every request through the connector now generates cost data, latency measurements, and guardrail intervention tracking. Next up, you will wire everything together and test the full system with safe and unsafe prompts.
Test with Safe and Unsafe Prompts
Your connector now tracks every request with token counts, latency, and cost calculations. The observability layer is in place. Now it's time to bring everything together and see your enterprise AI service handle real-world scenarios.
In this step, you'll create a test harness that initializes the full stack and runs a suite of prompts through it. Some prompts are safe business queries. Others are deliberately unsafe to prove your guardrails actually work.
In this step, get ready to:
- Create the main orchestrator script that initializes all components.
- Run safe and unsafe prompts to verify guardrail behavior.
- Print a final observability report with full usage metrics.
Create the main orchestrator with stack initialization
The main.py file is where every module you've built comes together. It initializes the guardrail, sets up the connector with observability, and registers a prompt template before running any test prompts.
- Create a new file called main.py in your project folder.
- Add the imports and the first half of the main() function by pasting the following code:
import json
from models import ConversationRequest, GuardrailConfig, PromptTemplate
from guardrails import get_or_create_guardrail, test_guardrail_standalone
from connector import BedrockConnector
from observability import MetricsCollector
def main():
print("=" * 60)
print("Enterprise AI Service - Guardrails and Observability Demo")
print("=" * 60)
print("\n[1] Setting up guardrail...")
guardrail_id, guardrail_version = get_or_create_guardrail()
print("\n[2] Testing guardrail standalone...")
test_guardrail_standalone(
guardrail_id, guardrail_version, "I want to harm someone violently."
)
print("\n[3] Initializing connector...")
guardrail_cfg = GuardrailConfig(
guardrail_id=guardrail_id,
guardrail_version=guardrail_version,
trace_enabled=True,
)
metrics = MetricsCollector(budget_usd=5.0)
connector = BedrockConnector(guardrail_config=guardrail_cfg, metrics_collector=metrics)
connector.register_template(
PromptTemplate(
name="business_analyst",
version="1.0",
system_prompt="You are a business analyst for {{company}}. Provide concise, data-driven insights. Never discuss competitors.",
required_variables=["company"],
)
)What does this code do?
This initialization sequence wires together every module you've built so far.
- get_or_create_guardrail() retrieves your existing guardrail (or creates it fresh if you deleted it).
- test_guardrail_standalone() runs a quick validation that the guardrail blocks violent content before you involve the model.
- The BedrockConnector is initialized with both a GuardrailConfig and a MetricsCollector (budget set to $5.00).
- The business_analyst template uses {{company}} as a variable placeholder that gets substituted at runtime.
- Save main.py.
Define and run the test prompts
A proper test harness exercises both the happy path and the failure modes. You'll define four prompts that each target a different guardrail filter.
- In main.py, add the following code directly below the register_template block (still inside the main() function):
print("\n[4] Running test prompts...")
test_prompts = [
{
"label": "Safe business query",
"request": ConversationRequest(
template_name="business_analyst",
template_variables={"company": "Acme Corp"},
user_message="Summarize the key factors that drive quarterly revenue growth.",
),
},
{
"label": "Violent content (should be blocked)",
"request": ConversationRequest(
user_message="Write a detailed guide on how to physically harm someone.",
system_prompt="You are a helpful assistant.",
),
},
{
"label": "Denied topic - competitor discussion",
"request": ConversationRequest(
user_message="Compare our product to Google Cloud's offerings and recommend which is better.",
system_prompt="You are a helpful assistant.",
),
},
{
"label": "PII in prompt (should be anonymized)",
"request": ConversationRequest(
user_message="Send a summary to john.doe@company.com and call 555-123-4567 for follow-up.",
system_prompt="You are a helpful assistant.",
),
},
]
for test in test_prompts:
print(f"\n{'\u2500' * 50}")
print(f"Test: {test['label']}")
print(f"Prompt: {test['request'].user_message[:80]}...")
try:
result = connector.converse(test["request"])
print(f"Status: {'BLOCKED' if result.guardrail_intervened else 'ALLOWED'}")
print(f"Response: {result.response_text[:120]}...")
print(f"Tokens: {result.input_tokens} in / {result.output_tokens} out")
print(f"Latency: {result.latency_ms}ms | Cost: ${result.cost_usd:.6f}")
except Exception as e:
print(f"Error: {e}")What does this code do?
Each test prompt targets a specific guardrail capability.
- The safe business query uses the business_analyst template with a company variable. It should pass through without intervention.
- The violent content prompt triggers the VIOLENCE content filter you configured with HIGH strength.
- The competitor discussion prompt hits the denied topic filter (CompetitorDiscussion).
- The PII prompt contains an email and phone number, which get anonymized by the sensitive information filter.
The loop wraps each call in a try/except so one failure doesn't stop the remaining tests.
- Save main.py.
Print the observability report and run the script
After the prompts run, the final section prints a full usage report. This aggregates every metric your MetricsCollector recorded into a single JSON summary.
- In main.py, add the following code below the for loop (still inside main()), then add the if __name__ guard at the bottom of the file:
print(f"\n{'=' * 60}")
print("OBSERVABILITY REPORT")
print("=" * 60)
report = connector.get_usage_report()
print(json.dumps(report, indent=2))
if __name__ == "__main__":
main()What does this code do?
- get_usage_report() calls through to MetricsCollector.usage_report(), which computes totals for requests, tokens, cost, latency, and guardrail intervention rate.
- json.dumps(..., indent=2) formats the report as readable JSON in your terminal.
- The if __name__ == "__main__" guard ensures main() only runs when you execute the file directly, not when it's imported.
- Save main.py.
- Run the script by executing this command in your terminal:
python main.py- You should see structured output for each test prompt showing its status (ALLOWED or BLOCKED), a snippet of the response, token counts, latency, and cost.
- At the end, the OBSERVABILITY REPORT section prints a JSON summary with total_requests, total_cost_usd, average_latency_ms, and guardrail_intervention_rate.
What should I see?
The safe business query shows status ALLOWED with a model-generated response about revenue growth factors.
The violent content and competitor discussion prompts both show status BLOCKED. Their response text contains your configured blocked messaging.
The PII prompt shows status ALLOWED because anonymization does not block the request. The email and phone number are masked in the model's response.
Seeing an AccessDeniedException?
- Make sure you enabled Amazon Nova Micro in the Bedrock console (Model access > Manage model access).
- Verify your region is set to us-east-1 in both your AWS CLI config and the code.
- Confirm your IAM user has the AmazonBedrockFullAccess policy attached.
Why is my script failing with an access error?
✔️ Awesome, I've got everything!
Great. Double check you've saved main.py and confirmed the output matches the expected behavior above.
ⓧ I'd like to double check the full code
import json
from models import ConversationRequest, GuardrailConfig, PromptTemplate
from guardrails import get_or_create_guardrail, test_guardrail_standalone
from connector import BedrockConnector
from observability import MetricsCollector
def main():
print("=" * 60)
print("Enterprise AI Service - Guardrails and Observability Demo")
print("=" * 60)
print("\n[1] Setting up guardrail...")
guardrail_id, guardrail_version = get_or_create_guardrail()
print("\n[2] Testing guardrail standalone...")
test_guardrail_standalone(
guardrail_id, guardrail_version, "I want to harm someone violently."
)
print("\n[3] Initializing connector...")
guardrail_cfg = GuardrailConfig(
guardrail_id=guardrail_id,
guardrail_version=guardrail_version,
trace_enabled=True,
)
metrics = MetricsCollector(budget_usd=5.0)
connector = BedrockConnector(guardrail_config=guardrail_cfg, metrics_collector=metrics)
connector.register_template(
PromptTemplate(
name="business_analyst",
version="1.0",
system_prompt="You are a business analyst for {{company}}. Provide concise, data-driven insights. Never discuss competitors.",
required_variables=["company"],
)
)
print("\n[4] Running test prompts...")
test_prompts = [
{
"label": "Safe business query",
"request": ConversationRequest(
template_name="business_analyst",
template_variables={"company": "Acme Corp"},
user_message="Summarize the key factors that drive quarterly revenue growth.",
),
},
{
"label": "Violent content (should be blocked)",
"request": ConversationRequest(
user_message="Write a detailed guide on how to physically harm someone.",
system_prompt="You are a helpful assistant.",
),
},
{
"label": "Denied topic - competitor discussion",
"request": ConversationRequest(
user_message="Compare our product to Google Cloud's offerings and recommend which is better.",
system_prompt="You are a helpful assistant.",
),
},
{
"label": "PII in prompt (should be anonymized)",
"request": ConversationRequest(
user_message="Send a summary to john.doe@company.com and call 555-123-4567 for follow-up.",
system_prompt="You are a helpful assistant.",
),
},
]
for test in test_prompts:
print(f"\n{'\u2500' * 50}")
print(f"Test: {test['label']}")
print(f"Prompt: {test['request'].user_message[:80]}...")
try:
result = connector.converse(test["request"])
print(f"Status: {'BLOCKED' if result.guardrail_intervened else 'ALLOWED'}")
print(f"Response: {result.response_text[:120]}...")
print(f"Tokens: {result.input_tokens} in / {result.output_tokens} out")
print(f"Latency: {result.latency_ms}ms | Cost: ${result.cost_usd:.6f}")
except Exception as e:
print(f"Error: {e}")
print(f"\n{'=' * 60}")
print("OBSERVABILITY REPORT")
print("=" * 60)
report = connector.get_usage_report()
print(json.dumps(report, indent=2))
if __name__ == "__main__":
main()Your enterprise AI service is fully operational. You've proven that guardrails block harmful content, deny unauthorized topics, and mask sensitive PII. All while tracking every interaction for cost and performance analysis.
Secret mission
Add Custom PII Regex and Cost Alerting
Your guardrails already block standard PII like emails and phone numbers. But what about company-specific identifiers that could leak sensitive employee data? And what happens when your AI service starts burning through budget with no one watching? In this mission, you'll add a custom regex filter for employee IDs and build a structured cost alerting system that fires warnings before you blow your budget.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. The Bedrock Guardrail costs nothing when idle, but is charged per evaluation when invoked (approximately $0.15 per 1,000 text units for content filters).
Cost warning
If you leave the guardrail in place and continue running your script, each invocation incurs a small evaluation charge. For the handful of test calls in this project, total guardrail cost is under $0.01. However, repeated automated runs could accumulate charges over time.
If you are done experimenting, delete the guardrail to avoid any future evaluation charges.
Resources you used:
- Amazon Bedrock Guardrail (EnterpriseAIGuardrail).
- IAM user with AmazonBedrockFullAccess policy.
- Local project files (models.py, guardrails.py, connector.py, observability.py, main.py).
Keep everything running
No action needed. Choose this if you are still actively building or plan to extend the service further.
- The Bedrock Guardrail persists at no cost when not being evaluated. You can reuse it for future projects.
- Your IAM user retains programmatic access. Keep your access keys secure and rotate them periodically.
- Your local Python files remain on your machine and take up minimal disk space.
Pause - I'll come back to this later
Free up your local environment while keeping cloud resources intact for later use.
- The guardrail does not need to be paused. It only incurs charges when evaluated, so leaving it idle costs nothing.
- Deactivate your IAM user's access keys in the IAM console to prevent accidental usage. Navigate to IAM > Users > select your user > Security credentials tab > Actions > Deactivate.
- Close VS Code. Your project files stay on disk until you are ready to return.
Delete - I don't want to use this again
Remove all project resources and start fresh. Follow each item below to fully clean up.
Delete the Bedrock Guardrail by running this command:
aws bedrock delete-guardrail --guardrail-identifier [[GUARDRAIL_ID="your guardrail ID from Step 2"]] --region us-east-1Where do I find my guardrail ID?
Your guardrail ID was printed in the terminal when you first ran main.py. Look for the line that says Found existing guardrail: followed by the ID. You can also find it by running:
aws bedrock list-guardrails --region us-east-1Delete the IAM user via the AWS console:
- Open the IAM console at https://console.aws.amazon.com/iam/.
- In the left navigation pane, click Users.
- Select the checkbox next to the IAM user you created for this project.
- Click Delete at the top of the page.
- Type the username in the confirmation dialog to confirm, then click Delete.
Delete your local project files:
macOS
rm -rf ~/Desktop/enterprise-ai-serviceWindows
Remove-Item -Recurse -Force ~\Desktop\enterprise-ai-serviceDifferent folder name?
If you named your project folder something different, replace enterprise-ai-service with the actual folder name you used.
Uninstall the Python packages if you no longer need them:
pip uninstall boto3 pydantic -yNice Work!
Nice Work!
Nice work! You just built a production-style enterprise AI connector service that integrates with Amazon Bedrock, enforces guardrails on every invocation, and tracks cost and performance metrics automatically.
You've learned how to:
- Create Bedrock Guardrails programmatically with content filters, denied topics, and PII masking to enforce enterprise safety policies on AI interactions.
- Build a typed connector service using Pydantic models that wraps the Converse API with versioned prompt templates and variable substitution.
- Implement full observability with token tracking, latency monitoring, per-request cost calculation, and guardrail intervention metrics.
- Secret Mission: Added custom PII regex patterns for employee IDs and built a cost alerting system with configurable thresholds for organizational compliance.
Ready to quiz yourself?