亚马逊Bedrock AgentCore网关策略与Lambda拦截器全面防护人工智能智能体安全实践

企业级AI Agent的安全治理，正在成为构建智能解决方案时绕不开的核心挑战。随着越来越多的企业开始用AI Agent来自动化工作流程，一个棘手的问题浮出水面：如何在组织内部，为成百上千个Agent安全地管理工具访问权限？

现代化的企业AI平台，往往运行着数百个Agent，服务于组织内的各类用户。这些Agent需要访问数千个跨团队、跨组织、甚至跨业务部门的MCP工具。这种规模本身就带来了一个根本性的治理难题——传统应用的执行逻辑是固定的，而基于大语言模型的Agent，需要在运行时自主决定调用哪个工具、传入什么参数、按什么顺序执行。由于工作流的动态特性，事先审计调用路径变得异常困难。你必须为LLM构建一套机制，确保它按照你的预期行事。

好消息是，Amazon Bedrock AgentCore网关提供了两种互补的机制来解决这个问题：Policy用于确定性访问控制，Interceptor用于动态验证。Policy让你可以在网关上定义针对工具的访问策略，这些策略用Cedar声明式策略语言编写，它会根据主体、操作、资源以及可选的请求上下文条件，对每个请求进行评判，并输出确定的“允许”或“拒绝”结果，同时自动记录到审计日志。而Lambda Interceptor则允许你编写自定义代码，在每次工具调用之前或之后执行，从而支持动态验证、负载增强、令牌交换和响应过滤。将这两种机制结合起来，就能为你的Agent方案构建出一套分层的安全架构。

接下来，我们以一个湖仓数据Agent为例，详细演示如何用Policy做确定性访问控制，用Lambda Interceptor做动态验证，以及如何组合两者来实现基于地理位置的访问控制——这个场景恰好同时需要动态验证和确定性决策。

前期准备

在开始之前，请确保你已经准备好：

• 一个AWS账户
• GitHub仓库的访问权限
• 配置前期资源所需的IAM权限

方案概览

这个湖仓数据Agent是一个AI助手，供保险公司员工查询理赔数据。数据存储在Amazon S3 Tables中，通过Amazon Athena和AWS Lake Formation进行查询。应用里定义了三种用户角色：投保人（只能查看自己的理赔）、理赔员（管理被分配的理赔）和管理员（拥有包括审计日志在内的完整数据访问权限）。前端通过Streamlit UI，由Amazon Cognito完成用户认证，并将JWT令牌传递给Agent。

MCP服务端暴露了五个工具：query_claims、get_claim_details、get_claims_summary、query_login_audit和text_to_sql。角色与工具的映射关系、租户IAM角色映射以及用户所在的地理位置都存储在DynamoDB中。AWS Lake Formation则在查询时强制执行行级和列级安全策略。这样一来，即使Agent构造了一个宽泛的SQL查询，结果也会自动被限定为调用者IAM角色所能看到的数据范围。

下图为湖仓数据Agent的架构示意图：

用户通过Streamlit UI访问湖仓Agent，Amazon Cognito负责认证并签发Bearer令牌。AgentCore Runtime承载湖仓Agent，验证这些令牌并为每个用户建立隔离会话。当Agent调用工具时，AgentCore网关将请求路由到Lambda Interceptor。Interceptor提取Bearer令牌，通过租户角色映射验证工具访问权限，然后生成一个包含租户范围声明的令牌。接着，AgentCore策略引擎在允许访问之前，根据定义的策略评估每个工具调用。最后，湖仓MCP服务端使用具备作用域限制的凭证查询数据。AWS Lake Formation基于用户表和理赔表强制执行行级和列级安全，确保每个用户只能看到自己有权访问的数据。AgentCore的可观测性和会话日志会流式传输到CloudWatch，用于实时监控和合规审计。

请求流程

下图展示了通过该方案的完整工具调用流程：

当湖仓Agent通过AgentCore网关发起工具调用时，请求首先被请求Interceptor Lambda函数截获。这个Interceptor负责转换请求——将Bearer令牌替换为具有租户作用域的凭证，并注入额外的上下文信息。然后，策略引擎会根据Cedar策略评估转换后的请求。评估通过后，转换后的请求才会被用于调用湖仓MCP服务端的工具。最后，响应Interceptor Lambda函数会在结果返回给用户之前，对工具列表进行过滤。

这里有一个关键点：网关在评估Cedar策略之前，会先执行请求Interceptor。这个顺序是整个设计模式的基础——你可以先利用Interceptor来丰富请求上下文，然后再用策略来评估这个已丰富的上下文。

AgentCore网关中的策略实施

Amazon Bedrock AgentCore中的Policy使用Cedar策略语言，在网关层面实施确定性的、可审计的访问控制。Cedar策略由permit或forbid规则组成，这些规则基于主体、操作和资源进行评估，并根据操作上下文附带条件。

当授权规则可以表达为对身份属性、操作标识符和请求上下文的逻辑条件时，Cedar策略就非常适合用于细粒度访问控制。典型的用例包括：限制特定角色可以调用哪些工具、阻止某些用户组访问敏感操作。此外，Cedar还能基于Interceptor注入的上下文属性来强制执行数据驻留规则，或者在请求到达下游服务之前在网关层进行作用域检查或时间窗口限制。

设计模式一：仅使用Policy

先来看一个将Policy作为湖仓Agent安全层的例子。假设业务部门决定：投保人不能调用get_claims_summary工具。投保人可以查看自己个人的理赔记录，但汇总统计数据只能由理赔员和管理员访问。为此，你只需要在网关上挂载一个策略引擎，然后定义两条协同工作的Cedar策略：一条基础的permit规则，一条有针对性的forbid规则。

当策略引擎挂载到网关上时，它遵循默认拒绝的语义——如果没有策略明确允许某个请求，该请求就会被拒绝。因此，你首先需要一条基础的permit策略，允许Agent调用网关上的工具：

permit(principal,action,resource == AgentCore::Gateway::"");

只有这条策略的话，所有经过认证的用户都能调用任何工具。

接下来，添加一条forbid规则来专门限制投保人。在Cedar中，forbid规则优先于permit规则，所以这一条规则就足以阻止对目标工具的调用，同时保留其他所有访问权限。

forbid(principal is AgentCore::OAuthUser,action == AgentCore::Action::"lakehouse-mcp-target___get_claims_summary",resource == AgentCore::Gateway::"") when {principal.hasTag("cognito:groups") &&principal.getTag("cognito:groups") like "*policyholders*"};

这两条策略组合起来的效果就是：Agent可以调用任何工具，但投保人试图访问理赔汇总数据时会被阻止。

注意：一个最佳实践是，先将策略引擎的策略执行模式设置为LOG_ONLY。在这种模式下，所有策略决策都会写入CloudWatch，但不会阻止任何请求。这样你就可以在切换到ENFORCE模式之前，验证每条策略规则的行为是否符合预期。

下图展示了仅使用Policy模式下的工具调用流程：

当湖仓Agent发送请求时，AgentCore网关首先使用内置的授权功能验证JWT令牌。然后，策略引擎根据挂载的Cedar策略组合来评估请求。在这个例子中，Cedar策略使用了forbid-permit模式：它首先禁止OAuth用户调用get_claims_summary工具，然后仅在主体具有匹配policyholders的Cognito组标签时才允许访问。这种确定性的策略评估确保了只有属于授权组的用户才能调用特定工具。根据策略评估的结果，网关要么允许调用湖仓MCP服务端并将原始响应返回给Agent，要么在请求到达工具之前就将其拒绝。

设计模式一的策略评估结果

基于策略的管控优势

Cedar策略为保护AI Agent提供了三个关键优势：

• 确定性——相同的输入总能产生相同的决策，不受LLM行为影响。
• 可审计性——只要为网关启用了CloudWatch日志投递，每次“允许”或“拒绝”的决策都会连同完整上下文一起被记录，提供完整的审计追踪。
• 低延迟——Cedar的评估过程对请求处理带来的开销微乎其微。

用于动态控制的Interceptor

Interceptor是AgentCore网关在请求生命周期的两个阶段调用的自定义Lambda函数。REQUEST拦截器在下游工具收到请求之前运行，而RESPONSE拦截器在响应返回给Agent之前运行。网关会向每个Interceptor传递一个包含mcp键的JSON事件，其中包含原始的请求头和请求体。Interceptor负责转换请求内容，并以相同结构返回。Interceptor适用于所有网关目标类型，包括Lambda函数、OpenAPI端点和MCP服务端。

当Agent代表用户调用工具时，一个关键的安全决策是：身份信息如何在调用链中传播。一种简单的方法是将原始用户的JWT原封不动地传递给每个下游服务。这种方法虽然简单，但会让下游服务获得超出其所需的权限。一旦某个服务被攻破，攻击者就可能在别处滥用这个权限过大的令牌。另一种方法是“袋里授权”模式，每个下游目标都会收到一个独立的、仅具有最低必要权限的令牌，专门为该服务而创建。用户的身份上下文则用于审计目的。设计模式二就实现了这种模式。REQUEST拦截器通过sts:AssumeRole将用户的Cognito JWT交换为短期有效的、具有租户作用域的IAM凭证，然后MCP服务端实际收到的就是这些具有作用域限制的凭证。

设计模式二：仅使用Interceptor——袋里授权的令牌交换与上下文传播

在REQUEST拦截器中，Cedar无法完成以下三个操作：

• JWT到IAM令牌的交换（袋里授权）。从JWT中读取用户的Cognito组，在DynamoDB中查找对应的租户IAM角色，然后调用sts:AssumeRole获取短期有效的作用域凭证。
• 上下文注入。将用户身份和临时IAM凭证写入MCP请求体的params.arguments.context中，这样MCP服务端就可以用它们来构造具有作用域限制的Athena客户端。
• 工具授权。在转发请求之前，检查DynamoDB中的allowed_tools，对于未授权的调用返回结构化的MCP错误。

简化后的REQUEST拦截处理器代码如下：

def lambda_handler(event, context):# 从拦截器事件中解析MCP网关请求mcp_data = event.get('mcp', {})gateway_request = mcp_data.get('gatewayRequest', {})body = gateway_request.get('body', {})headers = gateway_request.get('headers', {})token = extract_bearer_token(headers)claims = validate_and_decode_jwt(token)# 第一步：验证Cognito JWT# 第二步：根据DynamoDB中的allowed_tools检查工具授权is_authorized, error_msg, tool_name = validate_tool_access(claims, body)if not is_authorized:return build_mcp_error_response(error_msg, status_code=403)# 第三步：袋里授权——将JWT组声明交换为租户IAM凭证claim_name, claim_value = get_claim_for_exchange(claims)tenant_credentials = exchange_jwt_to_iam(claim_name, claim_value)# sts:AssumeRole# 第四步：将用户身份和作用域凭证注入MCP请求体if 'params' in body and 'arguments' in body['params']:body['params']['arguments']['context'] = {'user_id': user_principal,'tenant_credentials': {'access_key_id': tenant_credentials['AccessKeyId'],'secret_access_key': tenant_credentials['SecretAccessKey'],'session_token': tenant_credentials['SessionToken'],}}# 以要求的拦截器输出格式返回转换后的请求return {'interceptorOutputVersion': '1.0','mcp': {'transformedGatewayRequest': {'headers': transformed_headers,'body': body,}}}

MCP服务端收到的是包含注入上下文的转换后请求。每个工具函数都接收一个context参数，并用它来构造具有作用域限制的Athena客户端。然后，Lake Formation会根据租户角色的权限自动在查询时应用行级和列级过滤，无需手动添加SQL的WHERE子句：

# server.py --- query_claims工具def query_claims(claim_status=None, context=None):user_id, tenant_creds = get_user_id_with_fallback(context)# Athena客户端使用租户的作用域IAM凭证（而非用户的JWT）# Lake Formation自动应用行级和列级过滤athena_client = boto3.client('athena',aws_access_key_id=tenant_creds['access_key_id'],aws_secret_access_key=tenant_creds['secret_access_key'],aws_session_token=tenant_creds['session_token'])...

仅使用Interceptor模式的调用流程

下图展示了仅使用Interceptor模式的调用流程：

当湖仓Agent发送请求时，AgentCore网关验证JWT令牌，然后将原始请求作为包含mcp键的JSON事件路由到网关请求Interceptor Lambda。这个Interceptor会转换请求，将Cognito JWT交换为具有租户作用域的凭证，并验证工具授权。接着，网关使用转换后的请求（包含注入的上下文和租户凭证）调用湖仓MCP服务端。当MCP服务端返回原始响应时，网关响应Interceptor会在结果返回给Agent之前对其进行处理。这个Interceptor会动态过滤工具列表、根据用户权限编辑敏感信息，确保每个用户只能看到他们有权访问的工具和数据。

使用响应Interceptor进行动态工具过滤

响应Interceptor还让你能够控制Agent在工具响应后看到的内容。最常见的用途是过滤工具列表和语义搜索结果，让每个用户只看到他们有权调用的工具。你还可以将其与Amazon Bedrock Guardrails等服务集成，用于个人身份信息等敏感数据的脱敏。这样做既能通过隐藏未授权工具来增强安全性，防止敏感信息泄露，也能通过向LLM提供更小、更精确的工具列表来减少工具选择的错误，提升可靠性。

何时使用Policy，何时使用Lambda Interceptor

Policy和Interceptor并非可以互换。它们在安全架构中服务于不同的目的。下表总结了关键的决策标准。

使用Policy的场景：

• 需要一个无法被Agent或LLM绕过的、硬性的、可审计的边界。
• 授权规则仅依赖于身份声明、操作名称、资源ARN或请求中已有的上下文。
• 需要一个紧急“断流”开关——一条forbid规则通过控制面API就能立即生效。

使用Interceptor的场景：

• 规则需要运行时获取数据（DynamoDB、密钥、外部授权服务）。
• 需要在请求到达工具之前转换或丰富请求负载。
• 需要在响应返回给Agent之前过滤或清理工具响应。
• 授权决策是有状态的——例如令牌交换、每用户速率限制。
• 需要在方法级别（tools/call与tools/list）强制执行授权，而非工具级别。

设计目标在于组合使用。将一切本质上动态的任务交给Interceptor，将一切可以表达为针对丰富上下文的逻辑规则的任务交给Cedar。由于REQUEST拦截器在Cedar之前执行，这两种机制形成了一个自然的流水线，而不是争夺同一职责。

当Policy和Interceptor协同工作时，每一层都处理自己擅长的事情。下图展示了结合Policy和Lambda Interceptor的分层安全调用流程：

在此模式中，当湖仓Agent发送请求时，AgentCore网关验证JWT令牌，然后将原始请求路由到网关请求Interceptor Lambda。这个Interceptor通过动态注入geography、user_id和租户凭证来丰富请求。接着，策略引擎基于这个已经丰富的上下文执行确定性的Cedar策略评估，提供一致的访问决策。如果被允许，网关就使用包含注入租户凭证的转换后请求调用湖仓MCP服务端。当MCP服务端返回原始响应时，网关响应Interceptor会根据用户权限动态过滤工具列表并编辑敏感信息，然后将转换后的响应返回给Agent。

评估顺序是：REQUEST拦截器先于Cedar策略。通过这种组合，你可以先用Interceptor从任何数据源获取数据并注入到请求参数中，然后用Cedar策略来评估这个已经丰富后的请求。

设计模式三：Policy + Interceptor——基于地理位置的访问控制

这个模式解决的是一个典型的合规需求。我们想要创建一条边界：来自欧盟辖区的用户不应访问个人理赔记录，只能查看汇总数据。这是一个数据驻留规则，它结合了一个动态属性（存储在DynamoDB中的用户geography）和一个确定性策略规则（欧盟用户不能调用query_claims或get_claim_details）。

Cedar无法从DynamoDB获取geography，而Lambda Interceptor无法声明forbid语义并带有自动审计日志。但Policy和Lambda Interceptor的组合可以同时处理这两者：用Lambda Interceptor获取geography并丰富请求，然后Policy基于这个丰富后的请求，在将请求传递给目标之前，根据用户的地理位置评估个人理赔记录的访问权限。

第一步：Interceptor获取地理位置并将其注入工具参数

# interceptor-request/lambda_function.py# 生产环境中从DynamoDB表 'lakehouse_user_geography' 获取地理位置# 为简化演示，这里使用Lambda内部映射USER_GEOGRAPHY: Dict[str, str] = {'policyholder001@example.com': 'US','policyholder002@example.com': 'EU','adjuster001@example.com': 'US','admin@example.com': 'US',}# 在现有的上下文注入之后，将地理位置注入到参数的顶层# Cedar将其评估为 context.input.geography。# 如果放在 context (params.arguments.context.geography) 内部，# Cedar需要写成 context.input.context.geography --- 表达起来更复杂geography = USER_GEOGRAPHY.get(user_principal, 'UNKNOWN')if 'params' in transformed_body and 'arguments' in transformed_body['params']:transformed_body['params']['arguments']['geography'] = geographylogger.info(f'已为用户={user_principal} 注入 geography={geography}')

关键细节：Cedar将工具参数引用为context.input.。Cedar可以访问任何字段，无论嵌套深度如何，但将geography放在params.arguments的顶层可以使策略更简洁。这样，在策略中就可以直接引用为context.input.geography，而不是嵌套在内部的context.input.context.geography。

第二步：Cedar策略评估注入的地理位置

// 欧盟用户无法访问个人理赔记录（GDPR数据驻留要求）// 宽泛的 permit_all 规则仍允许欧盟用户调用 get_claims_summaryforbid(principal,action in [AgentCore::Action::"lakehouse-mcp-target___query_claims",AgentCore::Action::"lakehouse-mcp-target___get_claim_details"],resource == AgentCore::Gateway::"") when {context.input has geography &&context.input.geography == "EU"};// 受限制的地理位置，拒绝所有工具访问forbid(principal,action in [AgentCore::Action::"lakehouse-mcp-target___query_claims",AgentCore::Action::"lakehouse-mcp-target___get_claim_details",AgentCore::Action::"lakehouse-mcp-target___get_claims_summary",AgentCore::Action::"lakehouse-mcp-target___query_login_audit",AgentCore::Action::"lakehouse-mcp-target___text_to_sql"],resource == AgentCore::Gateway::"") when {context.input has geography &&context.input.geography == "RESTRICTED"};

所有三条forbid策略都由同一个Cedar策略引擎一并评估。只要有任意一条forbid规则匹配，请求就会被拒绝，无论是否有任何匹配的permit规则。

组合设计的职责矩阵

设计模式三的策略评估结果

端到端实施指南

要亲自尝试这个方案，首先克隆Amazon Bedrock AgentCore示例仓库，并进入湖仓Agent目录：

git clone https://github.com/awslabs/amazon-bedrock-agentcore-samples.gitcd amazon-bedrock-agentcore-samples/02-use-cases/lakehouse-agent

然后，按照该目录中README文件的设置和部署说明，配置你的AWS环境，并使用CLI脚本完成部署。

第一步：前置部署（生成cdk.json、分离Interceptor、更新Lambda）

为准备CDK部署，运行pre-deploy.sh脚本，一步完成以下操作：

• 从SSM参数存储自动生成cdk.json
• 临时将Interceptor从网关分离
• 更新并重新部署支持设计模式三的请求Interceptor Lambda函数

cd 02-use-cases/lakehouse-agent/cdkbash scripts/pre-deploy.sh

第二步：CDK部署

使用CDK创建策略引擎、创建四条Cedar策略，并将策略引擎和Interceptor挂载到AgentCore网关上。

# 安装npm依赖npm ci# 引导AWS账户（每个账户和区域仅需执行一次）# npx cdk boostrapnpx cdk deploy --require-approval never --profile

第三步：使用测试请求进行验证

使用policyholder002（geography=EU）的凭证调用Agent，确认query_claims会因欧盟geography的forbid规则返回403错误。然后验证get_claims_summary也会返回403，这是由设计模式一的投保人防护规则截获的。再用policyholder001（geography=US）进行测试，确认query_claims调用成功，并且只返回该用户自己的理赔记录（由AWS Lake Formation强制执行）。

可观测性：贯穿整个流水线的端到端可追溯性

AgentCore网关集成了AgentCore可观测性和Amazon CloudWatch，为每个强制执行层提供可追溯性。每个层都会留下独特的、可查询的追踪记录。网关JWT授权器会记录每次请求的令牌验证结果。REQUEST拦截器Lambda函数会记录JWT声明提取、DynamoDB查询结果、令牌交换结果以及geography注入信息。策略引擎会记录完整的授权上下文以及每次评估的ALLOW或DENY决策。RESPONSE拦截器Lambda函数会记录从tools/list和语义搜索中过滤了哪些工具，为每个用户的工具可见性提供记录。

后续步骤

所有三种设计模式的示例代码都已放在GitHub仓库中。建议从设计模式一演示的策略规则入手，然后随着你的安全和合规需求的增长，逐步构建设计模式二和模式三。

资源清理

建议清理你不再计划使用的任何资源，避免产生意外费用。在完成方案探索后，请按照说明进行清理。

总结

在这篇文章中，我们通过三种设计模式演示了如何使用Policy、Lambda Interceptor以及两者的组合来构建安全的Agent。当授权规则是确定性的，并且可以基于身份和上下文来表达时，使用Policy。当规则需要外部数据、负载转换或令牌交换时，使用Lambda Interceptor。当你需要在运行时获取动态上下文，并对其以声明式方式强制执行规则时，则将两者结合使用。你可以运用这些模式，在构建Agent方案时确保Agent行为的安全性。

来源：互联网