Amazon Bedrock Agent Guardrails with Veto
Add runtime tool authorization to Amazon Bedrock agents. Complement IAM with per-call policies on Lambda action groups. Block dangerous operations, enforce argument constraints, require human approval.
Bedrock agent authorization with Veto
Amazon Bedrock Agents use action groups backed by Lambda functions or return-of-control to execute tools. IAM policies control which AWS resources the Lambda can access, but they cannot inspect tool arguments or enforce business rules on individual calls. Veto adds a per-call authorization layer inside your Lambda handlers, evaluating every action against your policies before execution.
IAM is not enough
AWS IAM controls infrastructure access: which Lambda can invoke which service, which DynamoDB table the agent can read. This is necessary but insufficient. IAM cannot answer questions like: "Should this agent transfer $50,000?" or "Can this agent send data to an external email?" These are runtime authorization decisions that depend on the arguments of each individual tool call.
IAM: infrastructure access
IAM controls which AWS services your Lambda can call. It answers "can this function invoke DynamoDB?" -- a binary yes/no at the service level. It cannot inspect the contents of the request.
Veto: runtime authorization
Veto controls what the agent does with its access. It answers "should this specific transfer of $50,000 to account EXT_789 execute?" -- inspecting arguments, applying business rules, enforcing rate limits.
Privilege escalation risk
Bedrock agents with broad Lambda execution roles can chain tool calls in unintended ways. An agent with sts:AssumeRole and lambda:InvokeFunction access could escalate privileges through creative tool chaining that IAM alone cannot prevent.
Data exfiltration via responses
Agents with database read access can return sensitive data in their text responses, bypassing data loss prevention controls. Veto can restrict which queries execute and what data patterns are allowed in results.
Lambda action group handler
Add Veto authorization inside your Lambda handler. When Bedrock invokes the Lambda with an action group request, Veto evaluates the action and arguments against your policies before your business logic executes.
import json
from veto import Veto
veto = None
async def get_veto():
global veto
if veto is None:
veto = await Veto.init()
return veto
def lambda_handler(event, context):
"""Bedrock Agent action group Lambda handler with Veto authorization."""
import asyncio
return asyncio.get_event_loop().run_until_complete(
_handle(event, context)
)
async def _handle(event, context):
v = await get_veto()
action_group = event.get("actionGroup", "")
api_path = event.get("apiPath", "")
http_method = event.get("httpMethod", "")
parameters = event.get("parameters", [])
request_body = event.get("requestBody", {})
# Convert Bedrock parameters to a dict for policy evaluation
args = {p["name"]: p["value"] for p in parameters}
if request_body:
body_content = request_body.get("content", {})
for content_type, props in body_content.items():
for prop in props.get("properties", []):
args[prop["name"]] = prop["value"]
tool_name = f"{action_group}/{api_path}"
# Veto evaluates the action against your policies
decision = await v.guard(tool_name, args)
if decision.decision == "deny":
return {
"messageVersion": "1.0",
"response": {
"actionGroup": action_group,
"apiPath": api_path,
"httpMethod": http_method,
"httpStatusCode": 403,
"responseBody": {
"application/json": {
"body": json.dumps({
"error": "Action blocked by policy",
"reason": decision.reason,
})
}
},
},
}
if decision.decision == "require_approval":
return {
"messageVersion": "1.0",
"response": {
"actionGroup": action_group,
"apiPath": api_path,
"httpMethod": http_method,
"httpStatusCode": 202,
"responseBody": {
"application/json": {
"body": json.dumps({
"status": "pending_approval",
"approval_id": decision.approval_id,
})
}
},
},
}
# Execute the actual business logic
result = execute_action(action_group, api_path, args)
return {
"messageVersion": "1.0",
"response": {
"actionGroup": action_group,
"apiPath": api_path,
"httpMethod": http_method,
"httpStatusCode": 200,
"responseBody": {
"application/json": {
"body": json.dumps(result)
}
},
},
}Return of control pattern
When using Bedrock's return-of-control mode, the agent returns tool call requests to your application instead of invoking Lambda directly. This gives you a natural interception point for Veto authorization before you execute the tool and send the result back.
import boto3
import json
from veto import Veto
bedrock_runtime = boto3.client("bedrock-agent-runtime")
veto = await Veto.init()
def invoke_agent_with_veto(agent_id, agent_alias_id, session_id, prompt):
"""Invoke a Bedrock agent with return-of-control and Veto authorization."""
response = bedrock_runtime.invoke_agent(
agentId=agent_id,
agentAliasId=agent_alias_id,
sessionId=session_id,
inputText=prompt,
)
for event in response["completion"]:
if "returnControl" in event:
invocation = event["returnControl"]["invocationInputs"][0]
action = invocation["functionInvocationInput"]
tool_name = action["function"]
args = {
p["name"]: p["value"]
for p in action.get("parameters", [])
}
# Veto authorization check
import asyncio
decision = asyncio.run(veto.guard(tool_name, args))
if decision.decision == "deny":
# Return error to the agent
send_result(
agent_id, agent_alias_id, session_id,
invocation["invocationId"],
{"error": decision.reason}
)
continue
# Execute and return result
result = execute_tool(tool_name, args)
send_result(
agent_id, agent_alias_id, session_id,
invocation["invocationId"],
result
)Powertools for AWS Lambda
If you use Powertools for AWS Lambda, add Veto guard checks inside your tool functions. The BedrockAgentResolver handles request/response formatting while Veto handles authorization.
from aws_lambda_powertools.event_handler import BedrockAgentResolver
from veto import Veto
app = BedrockAgentResolver()
veto = None
@app.tool(name="transfer_funds", description="Transfer money between accounts")
async def transfer_funds(
amount: float,
from_account: str,
to_account: str,
) -> dict:
# Initialize Veto lazily
global veto
if veto is None:
veto = await Veto.init()
decision = await veto.guard("transfer_funds", {
"amount": amount,
"from_account": from_account,
"to_account": to_account,
})
if decision.decision == "deny":
return {"error": decision.reason}
if decision.decision == "require_approval":
return {
"status": "pending_approval",
"approval_id": decision.approval_id,
}
# Execute transfer
return perform_transfer(amount, from_account, to_account)
@app.tool(name="query_database", description="Query the analytics database")
async def query_database(sql: str, database: str = "analytics") -> dict:
global veto
if veto is None:
veto = await Veto.init()
decision = await veto.guard("query_database", {
"sql": sql,
"database": database,
})
if decision.decision == "deny":
return {"error": decision.reason}
return execute_query(sql, database)
def lambda_handler(event, context):
return app.resolve(event, context)Policy rules
Define policies that complement your IAM roles. Veto rules inspect tool arguments, enforce business constraints, and rate-limit agent actions.
version: "1.0"
name: Bedrock agent policies
rules:
- id: block-production-writes
action: block
tools: ["*/POST*", "*/PUT*", "*/DELETE*"]
conditions:
- field: arguments.database
operator: equals
value: "production"
message: "Production writes blocked. Use staging."
- id: approve-large-transfers
action: require_approval
tools: [transfer_funds]
conditions:
- field: arguments.amount
operator: greater_than
value: 5000
message: "Transfers over $5000 require human approval"
- id: block-sql-mutations
action: block
tools: [query_database]
conditions:
- field: arguments.sql
operator: matches
value: "(?i)(DROP|DELETE|TRUNCATE|ALTER|UPDATE|INSERT)"
message: "Only SELECT queries allowed"
- id: restrict-cross-account
action: block
tools: [transfer_funds]
conditions:
- field: arguments.to_account
operator: matches
value: "^EXT_"
message: "External account transfers blocked"
- id: rate-limit-agent
action: block
tools: ["*"]
conditions:
- field: context.hourly_count
operator: greater_than
value: 100
message: "Agent hourly action limit exceeded"How Veto protects Bedrock agents
Agent decides to act
The Bedrock agent processes user input and decides to invoke an action group. It selects the API path and constructs parameters from the conversation.
Lambda receives the request
Bedrock invokes your Lambda function with the action group, API path, HTTP method, and parameters. Or in return-of-control mode, your application receives the invocation request.
Veto evaluates the action
Before your business logic runs, Veto evaluates the tool name and arguments against your policies. Deterministic rules check argument values, patterns, rate limits, and business constraints.
Enforcement
Allowed actions execute normally. Blocked actions return a 403 with a policy reason. The Bedrock agent receives the response and can adjust its approach. All decisions are logged.
Frequently asked questions
How is Veto different from Bedrock Guardrails?
Does Veto replace IAM policies?
How do I deploy Veto in a Lambda function?
Does Veto work with Bedrock's return-of-control mode?
Can I use Veto with Bedrock AgentCore?
Related integrations
Anthropic Claude agent guardrails
GeminiGoogle Gemini function calling guardrails
Python SDKNative Python SDK for agent authorization
Secure your Bedrock agents beyond IAM.