Python Agent
生成用于构建带工具的代理的 Python AI 代理,并可选择将其部署到 Amazon Bedrock AgentCore Runtime。使用 framework 选项选择代理框架:Strands(默认)或 LangChain(基于 LangGraph 构建)。
生成器通过服务器 protocol 公开您的代理。两个框架都支持 HTTP(默认)、用于与其他 A2A 兼容代理互操作的 Agent-to-Agent (A2A) 协议,以及用于通过 CopilotKit 直接前端集成的 AG-UI 协议。
您可以通过两种方式生成 Python Agent:
pnpm nx g @aws/nx-plugin:py#agentyarn nx g @aws/nx-plugin:py#agentnpx nx g @aws/nx-plugin:py#agentbunx nx g @aws/nx-plugin:py#agent- 安装 Nx Console VSCode Plugin 如果您尚未安装
- 在VSCode中打开Nx控制台
- 点击
Generate (UI)在"Common Nx Commands"部分 - 搜索
@aws/nx-plugin - py#agent - 填写必需参数
- 点击
Generate
| 参数 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| project 必需 | string | - | 要添加 Agent 的项目 |
| framework | strands | langchain | strands | 要使用的 Agent SDK。 |
| name | string | - | Agent 的名称(默认:agent) |
| auth | iam | cognito | iam | 用于对 Agent 进行身份验证的方法。仅在设置了 infra 时适用(当 infra 为 none 时忽略)。 |
| protocol | http | a2a | ag-ui | http | Agent 的服务器协议。HTTP 暴露一个 FastAPI HTTP 服务器。A2A 暴露一个 Agent-to-Agent 协议服务器。AG-UI 暴露一个 Agent-User Interaction 协议服务器,用于直接前端集成。 |
| iac | inherit | cdk | terraform | inherit | 首选的 IaC 提供商。默认情况下,这继承自您的初始选择。 |
| infra | agentcore | none | agentcore | 用于托管 Agent 的基础设施类型。 |
| session | s3 | dynamodb-s3 | in-memory | s3 | 用于持久化 Agent 会话的存储。LangChain 支持 's3' 或 'dynamodb-s3';Strands 支持 's3';'in-memory' 两者均适用。 |
| preferInstallDependencies | boolean | true | 是否在生成器运行后优先安装依赖项。设置为 false 可在批量运行多个生成器时延迟安装(如果后续生成器需要计算 Nx 项目图,仍会运行安装);在最后统一安装一次。 |
生成器将向您现有的 Python 项目添加以下文件。生成的文件取决于所选的 protocol:
HTTP 协议(默认)
Section titled “HTTP 协议(默认)”文件夹your-project/
文件夹your_module/
文件夹agent/ (or custom name if specified)
- __init__.py Python package initialization
- init.py FastAPI application setup with CORS and error handling middleware
- agent.py Main agent definition with sample tools
- session.py Resolves the framework-specific session persistence implementation
- main.py FastAPI entry point for Bedrock AgentCore Runtime
- Dockerfile Entry point for hosting your agent (excluded when
infrais set toNone)
- pyproject.toml Updated with Strands dependencies
- project.json Updated with agent serve targets
A2A 协议
Section titled “A2A 协议”入口点通过 A2A 协议公开您的代理(Strands 使用 Strands A2A Server;LangChain 将图包装在 a2a-sdk 执行器中),挂载到 FastAPI 应用程序上:
文件夹your-project/
文件夹your_module/
文件夹agent/ (or custom name if specified)
- __init__.py Python package initialization
- agent.py Main agent definition with sample tools
- session.py Resolves the framework-specific session persistence implementation
- main.py A2A server entry point
- Dockerfile Entry point for hosting your agent (excluded when
infrais set toNone)
- pyproject.toml Updated with framework and A2A dependencies
- project.json Updated with agent serve targets
AG-UI 协议
Section titled “AG-UI 协议”入口点通过 AG-UI 协议公开您的代理,以便与 CopilotKit 直接前端集成。Strands 代理使用 ag-ui-strands 集成;LangChain 代理使用 ag-ui-langgraph:
文件夹your-project/
文件夹your_module/
文件夹agent/ (or custom name if specified)
- __init__.py Python package initialization
- agent.py Main agent definition with sample tools
- session.py Resolves the framework-specific session persistence implementation
- main.py AG-UI server entry point
- Dockerfile Entry point for hosting your agent (excluded when
infrais set toNone)
- pyproject.toml Updated with framework and AG-UI dependencies
- project.json Updated with agent serve targets
由于此生成器根据您选择的 iac 提供基础设施即代码,它将在 packages/common 中创建一个项目,其中包含相关的 CDK 构造或 Terraform 模块。
通用基础设施即代码项目的结构如下:
文件夹packages/common/constructs
文件夹src
文件夹app/ Constructs for infrastructure specific to a project/generator
- …
文件夹core/ Generic constructs which are reused by constructs in
app- …
- index.ts Entry point exporting constructs from
app
- project.json Project build targets and configuration
文件夹packages/common/terraform
文件夹src
文件夹app/ Terraform modules for infrastructure specific to a project/generator
- …
文件夹core/ Generic modules which are reused by modules in
app- …
- project.json Project build targets and configuration
为了部署您的 Agent,会生成以下文件:
文件夹packages/common/constructs/src
文件夹app
文件夹agents
文件夹<project-name>
- <project-name>.ts CDK construct for deploying your agent
文件夹packages/common/terraform/src
文件夹app
文件夹agents
文件夹<project-name>
- <project-name>.tf Module for deploying your agent
文件夹core
文件夹agent-core
- runtime.tf Generic module for deploying to Bedrock AgentCore Runtime
如果您为 infra 选择了 none,则不会生成 CDK 构造或 Terraform 模块 — Agent 只能在本地运行。在此模式下,auth 选项将被忽略,因为没有托管端点需要进行身份验证。
当部署到 Bedrock AgentCore Runtime 时,agent 会被构建为容器镜像,推送到 Amazon ECR 并在 AgentCore Runtime 中运行。客户端调用 AgentCore Runtime 数据平面端点,该端点将请求转发到您的 agent。agent 调用 Amazon Bedrock 进行模型推理,并可能调用工具、MCP 服务器或下游 API。
使用 infra: none 时,不会生成 AWS 基础设施。agent 作为本地进程运行,并调用 Amazon Bedrock 进行模型推理。
使用您的代理
Section titled “使用您的代理”您可以编辑 agent.py 来添加工具、配置模型和自定义系统提示。API 取决于您选择的框架。
工具是 AI 代理可以调用以执行操作的函数。两个框架都使用基于装饰器的方法来定义工具,从函数名称和文档字符串派生工具名称和描述,并从您的类型提示生成输入模式。
from strands import Agent, tool
@tooldef calculate_sum(numbers: list[int]) -> int: """Calculate the sum of a list of numbers""" return sum(numbers)
@tooldef get_weather(city: str) -> str: """Get weather information for a city""" # Your weather API integration here return f"Weather in {city}: Sunny, 25°C"
# Add tools to your agentagent = Agent( system_prompt="You are a helpful assistant with access to various tools.", tools=[calculate_sum, get_weather],)from langchain.agents import create_agentfrom langchain_aws import ChatBedrockConversefrom langchain_core.tools import tool
@tooldef calculate_sum(numbers: list[int]) -> int: """Calculate the sum of a list of numbers""" return sum(numbers)
@tooldef get_weather(city: str) -> str: """Get weather information for a city""" # Your weather API integration here return f"Weather in {city}: Sunny, 25°C"
# Add tools to your agentagent = create_agent( model=ChatBedrockConverse(model=MODEL_ID, region_name=REGION), tools=[calculate_sum, get_weather], system_prompt="You are a helpful assistant with access to various tools.",)使用预构建工具
Section titled “使用预构建工具”Strands 通过 strands-tools 包提供了一系列预构建工具:
from strands_tools import current_time, http_request, file_read
agent = Agent( system_prompt="You are a helpful assistant.", tools=[current_time, http_request, file_read],)LangChain 提供了一个庞大的工具和集成生态系统。安装相关的集成包,然后将工具传递给 create_agent:
from langchain_community.tools import DuckDuckGoSearchRun
agent = create_agent( model=ChatBedrockConverse(model=MODEL_ID, region_name=REGION), tools=[DuckDuckGoSearchRun()], system_prompt="You are a helpful assistant.",)默认情况下,Strands 代理使用 Claude 4 Sonnet,但您可以自定义模型提供商。有关配置选项,请参阅 Strands 关于模型提供商的文档:
from strands import Agentfrom strands.models import BedrockModel
# Create a BedrockModelbedrock_model = BedrockModel( model_id="anthropic.claude-sonnet-4-20250514-v1:0", region_name="us-west-2", temperature=0.3,)
agent = Agent(model=bedrock_model)LangChain 代理使用 ChatBedrockConverse 模型。生成的代理从 MODEL_ID 和 AWS_REGION 环境变量读取模型 ID 和区域,但您可以直接在 agent.py 中配置模型:
from langchain_aws import ChatBedrockConverse
model = ChatBedrockConverse( model="anthropic.claude-sonnet-4-20250514-v1:0", region_name="us-west-2", temperature=0.3,)使用 MCP 服务器
Section titled “使用 MCP 服务器”要使用您使用 py#mcp-server 或 ts#mcp-server 生成器创建的 MCP 服务器,您可以使用 connection 生成器,它将 MCP 服务器的工具连接到您的代理,适用于两个框架。
pnpm nx g @aws/nx-plugin:connectionyarn nx g @aws/nx-plugin:connectionnpx nx g @aws/nx-plugin:connectionbunx nx g @aws/nx-plugin:connection- 安装 Nx Console VSCode Plugin 如果您尚未安装
- 在VSCode中打开Nx控制台
- 点击
Generate (UI)在"Common Nx Commands"部分 - 搜索
@aws/nx-plugin - connection - 填写必需参数
- 点击
Generate
有关如何设置连接的详细信息,请参阅 connection 生成器指南。
对于其他 MCP 服务器,请参阅 Strands 或 LangChain MCP 文档。
有关编写代理的更深入指南,请参阅 Strands 或 LangChain 文档。
您的代理的服务器协议决定了它如何通信。所有选项都由 FastAPI 提供服务 — 入口点不同:
- HTTP(默认):具有自定义
/invocations端点、CORS 和流式传输的标准 FastAPI 服务器。最适合自定义客户端集成。 - A2A:挂载到 FastAPI 应用程序上的 Agent-to-Agent 服务器(Strands 使用 Strands A2A Server;LangChain 使用与框架无关的
a2a-sdk)。当您的代理需要被其他 A2A 兼容代理发现和调用时最佳。 - AG-UI:通过 SSE 的 AG-UI 协议(Strands 使用
ag-ui-strands;LangChain 使用ag-ui-langgraph)。最适合在 React 网站中与 CopilotKit 直接前端集成。
服务器入口点因框架而异(Strands 产生一个上下文管理的 Agent,而 LangChain 驱动一个编译的 create_agent 图),但每个协议的外部契约是相同的。
所有协议都公开 /ping 用于 AgentCore 运行时健康检查契约。A2A 代理监听端口 9000;HTTP 和 AG-UI 代理监听端口 8080。生成的 Dockerfile 和基础设施已为您配置。
FastAPI 服务器(HTTP 协议)
Section titled “FastAPI 服务器(HTTP 协议)”生成的 HTTP 服务器包括:
- 带有 CORS 中间件的 FastAPI 应用程序设置
- 错误处理中间件
- OpenAPI 模式生成
- 健康检查端点(
/ping) - 代理调用端点(
/invocations)
使用 Pydantic 自定义调用输入和输出
Section titled “使用 Pydantic 自定义调用输入和输出”代理的调用端点使用 Pydantic 模型来定义和验证请求和响应模式。您可以在 main.py 中自定义这些模型以匹配您的代理的要求。
定义输入模型
Section titled “定义输入模型”默认的 InvokeInput 模型接受一个提示。
from pydantic import BaseModel
class InvokeInput(BaseModel): prompt: str您可以扩展此模型以包含代理需要的任何其他字段。
会话 ID 从 x-amzn-bedrock-agentcore-runtime-session-id HTTP 标头中提取,与 Bedrock AgentCore Runtime 会话契约一致。如果未提供标头,则生成一个随机 UUID 作为后备。
定义输出模型
Section titled “定义输出模型”对于流式响应,生成器提供了 JsonStreamingResponse,它会自动将 Pydantic 模型序列化为 JSON Lines 格式(application/jsonl)。此格式与 OpenAPI 3.2 的流式规范兼容,并与生成的 TypeScript 客户端无缝协作。
默认情况下,代理产生包含代理响应文本的 StreamChunk 对象:
class StreamChunk(BaseModel): content: str您可以自定义 StreamChunk 模型以满足您的需求:
from pydantic import BaseModel
class StreamChunk(BaseModel): content: str timestamp: str token_count: intFastAPI 中有一个原生支持的开放功能请求。
Bedrock AgentCore Python SDK
Section titled “Bedrock AgentCore Python SDK”生成器包含对 Bedrock AgentCore Python SDK 的依赖,用于 PingStatus 常量。如果需要,可以直接使用 BedrockAgentCoreApp 而不是 FastAPI,但请注意会失去类型安全性。
您可以在此处的文档中找到有关 SDK 功能的更多详细信息。
A2A 服务器(A2A 协议)
Section titled “A2A 服务器(A2A 协议)”生成的 main.py 将 A2A 服务器挂载到也公开 /ping 的父 FastAPI 应用程序上。Strands 代理使用 Strands A2AServer;LangChain 代理将编译的图包装在 a2a-sdk AgentExecutor 中。当部署到 AgentCore 时,入口点从 AppConfig 解析运行时的公共 ARN 并在代理卡中公布它。
大多数用户不需要修改此文件;编辑 agent.py 以更改工具或系统提示。A2A 服务器从代理的 name 和 description 填充代理卡(/.well-known/agent-card.json)。
AG-UI 服务器(AG-UI 协议)
Section titled “AG-UI 服务器(AG-UI 协议)”生成的 main.py 公开了一个通过服务器发送事件(SSE)流式传输 AG-UI 事件的单个 POST 端点,以及用于 AgentCore 运行时健康检查的 /ping。连接取决于框架:
- Strands:将您的
Agent包装在ag_ui_strands.StrandsAgent中,在 FastAPIlifespan处理程序内构建(因此构建发生在容器/会话启动时而不是导入时),并从手工编写的 FastAPI/invocations循环中提供服务。 - LangChain:将编译的图包装在
ag_ui_langgraph.LangGraphAgent中,以相同的方式在lifespan内构建,并从手工编写的 FastAPI/invocations循环中提供服务。
大多数用户不需要修改此文件 — 编辑 agent.py 以更改工具或系统提示。
运行您的代理
Section titled “运行您的代理”要在本地运行您的 Agent(以及连接到它的所有内容),请使用项目的 dev 目标:
pnpm nx dev your-projectyarn nx dev your-projectnpx nx dev your-projectbunx nx dev your-project如果您已向项目添加了多个组件(代理、MCP 服务器等),这将启动它们全部。要仅运行此代理,请定位其 <your-agent-name>-dev 目标:
pnpm nx agent-dev your-projectyarn nx agent-dev your-projectnpx nx agent-dev your-projectbunx nx agent-dev your-project这使用 uv run 使用 Bedrock AgentCore Python SDK 执行您的 Agent。
与您的代理聊天
Section titled “与您的代理聊天”生成器配置了一个 <your-agent-name>-chat Nx 目标,可让您进入与代理的交互式终端聊天。
聊天目标独立运行。默认情况下,它连接到您本地运行的代理,因此首先启动代理的 <your-agent-name>-dev 目标(在单独的终端中):
pnpm nx agent-dev your-projectyarn nx agent-dev your-projectnpx nx agent-dev your-projectbunx nx agent-dev your-project然后,在另一个终端中,启动聊天:
pnpm nx run your-project:agent-chatyarn nx run your-project:agent-chatnpx nx run your-project:agent-chatbunx nx run your-project:agent-chat生成器为每个协议生成一个 scripts/<your-agent-name>/chat.ts。默认情况下,它连接到本地代理,或者在设置 RUNTIME_CONFIG_APP_ID 时连接到您部署的代理(请参阅下面的与您部署的代理聊天)。
对于 HTTP 代理,聊天脚本使用从代理的 OpenAPI 规范生成的类型安全 TypeScript 客户端。生成器还生成:
scripts/<your-agent-name>_openapi.py— 一个导出代理的 OpenAPI 规范的小脚本- 一个运行它的
<your-agent-name>-openapiNx 目标 - 一个在
scripts/<your-agent-name>/generated/下生成类型安全 TypeScript 客户端的<your-agent-name>-generate-clientNx 目标
当您自定义代理的输入形状(例如向 InvokeInput 添加新字段)时,更新 chat.ts 以在调用代理时传递新字段,其余部分将自动工作。
与您部署的代理聊天
Section titled “与您部署的代理聊天”要与部署到 Bedrock AgentCore 的代理聊天,请将 RUNTIME_CONFIG_APP_ID 环境变量设置为部署的 AppConfig 应用程序 ID(由部署的堆栈输出为 RuntimeConfigApplicationId)。聊天脚本从运行时配置解析您的代理的运行时 ARN 并连接到部署的端点:
对于 IAM 身份验证的代理,请求使用您的默认 AWS 凭证通过 SigV4 签名。确保环境具有有权调用运行时的 AWS 凭证:
RUNTIME_CONFIG_APP_ID=<app-id> pnpm nx run your-project:agent-chatRUNTIME_CONFIG_APP_ID=<app-id> yarn nx run your-project:agent-chatRUNTIME_CONFIG_APP_ID=<app-id> npx nx run your-project:agent-chatRUNTIME_CONFIG_APP_ID=<app-id> bunx nx run your-project:agent-chat对于 Cognito 身份验证的代理,通过 AGENT_ACCESS_TOKEN 环境变量提供 Cognito 访问令牌,该令牌作为承载令牌发送:
RUNTIME_CONFIG_APP_ID=<app-id> AGENT_ACCESS_TOKEN=<access-token> pnpm nx run your-project:agent-chatRUNTIME_CONFIG_APP_ID=<app-id> AGENT_ACCESS_TOKEN=<access-token> yarn nx run your-project:agent-chatRUNTIME_CONFIG_APP_ID=<app-id> AGENT_ACCESS_TOKEN=<access-token> npx nx run your-project:agent-chatRUNTIME_CONFIG_APP_ID=<app-id> AGENT_ACCESS_TOKEN=<access-token> bunx nx run your-project:agent-chat您可以使用 AWS CLI 的 cognito-idp admin-initiate-auth 命令获取访问令牌,例如:
aws cognito-idp admin-initiate-auth \ --user-pool-id <user-pool-id> \ --client-id <user-pool-client-id> \ --auth-flow ADMIN_NO_SRP_AUTH \ --auth-parameters USERNAME=<username>,PASSWORD=<password> \ --query 'AuthenticationResult.AccessToken' \ --output text将您的代理部署到 Bedrock AgentCore Runtime
Section titled “将您的代理部署到 Bedrock AgentCore Runtime”基础设施即代码
Section titled “基础设施即代码”如果您为 infra 选择了 agentcore,将生成相关的 CDK 或 Terraform 基础设施,您可以使用它将您的 Agent 部署到 Amazon Bedrock AgentCore Runtime。
将为您的 agent 生成一个 CDK 构造,根据您运行生成器时选择的 name 命名,或默认为 <ProjectName>Agent。
您可以在 CDK 应用程序中使用此 CDK 构造:
import { MyProjectAgent } from '@my-scope/common-constructs';
export class ExampleStack extends Stack { constructor(scope: Construct, id: string) { new MyProjectAgent(this, 'MyProjectAgent'); }}将为您生成一个 Terraform 模块,根据您运行生成器时选择的 name 命名,或默认为 <ProjectName>-agent。
将共享的 runtime_config_appconfig 模块的输出传递给 agent 模块:
module "my_project_agent" { source = "../../common/terraform/src/app/agents/my-project-agent"
appconfig_application_id = module.runtime_config_appconfig.application_id appconfig_application_arn = module.runtime_config_appconfig.application_arn}生成器提供了一个 auth 选项来配置您的 Agent 的身份验证。在生成 agent 时,您可以选择 IAM(默认)或 Cognito 身份验证。
默认情况下,您的 Agent 将使用 IAM 身份验证进行保护,只需在不带任何参数的情况下部署它:
import { MyProjectAgent } from '@my-scope/common-constructs';
export class ExampleStack extends Stack { constructor(scope: Construct, id: string) { new MyProjectAgent(this, 'MyProjectAgent'); }}您可以使用 grantInvokeAccess 方法授予在 Bedrock AgentCore Runtime 上调用您的 agent 的访问权限,例如:
import { MyProjectAgent } from '@my-scope/common-constructs';
export class ExampleStack extends Stack { constructor(scope: Construct, id: string) { const agent = new MyProjectAgent(this, 'MyProjectAgent'); const lambdaFunction = new Function(this, ...);
agent.grantInvokeAccess(lambdaFunction); }}# Agentmodule "my_project_agent" { # Relative path to the generated module in the common/terraform project source = "../../common/terraform/src/app/agents/my-project-agent"
appconfig_application_id = module.runtime_config_appconfig.application_id appconfig_application_arn = module.runtime_config_appconfig.application_arn}要授予调用您的 agent 的访问权限,您需要添加如下策略,引用 module.my_project_agent.agent_core_runtime_arn 输出:
{ Effect = "Allow" Action = [ "bedrock-agentcore:InvokeAgentRuntime" ] Resource = [ module.my_project_agent.agent_core_runtime_arn, "${module.my_project_agent.agent_core_runtime_arn}/*" ]}Cognito 身份验证
Section titled “Cognito 身份验证”当您选择 Cognito 身份验证时,生成器会将 agent 配置为使用 Cognito 进行身份验证。
生成的构造接受一个 identity 属性,用于配置 Cognito 身份验证:
import { MyProjectAgent, UserIdentity } from '@my-scope/common-constructs';
export class ExampleStack extends Stack { constructor(scope: Construct, id: string) { const identity = new UserIdentity(this, 'Identity');
new MyProjectAgent(this, 'MyProjectAgent', { identity, }); }}UserIdentity 构造可以使用 ts#website#auth 生成器生成,或者您可以创建自己的 CDK UserPool 和 UserPoolClient。
生成的模块接受 user_pool_id 和 user_pool_client_ids 变量用于 Cognito 身份验证:
module "user_identity" { source = "../../common/terraform/src/core/user-identity"}
module "my_project_agent" { source = "../../common/terraform/src/app/agents/my-project-agent"
appconfig_application_id = module.runtime_config_appconfig.application_id appconfig_application_arn = module.runtime_config_appconfig.application_arn
user_pool_id = module.user_identity.user_pool_id user_pool_client_ids = [module.user_identity.user_pool_client_id]}打包和 Docker 目标
Section titled “打包和 Docker 目标”为了为 Bedrock AgentCore Runtime 构建您的 Agent,会向您的项目添加一个 bundle 目标,它:
- 使用
uv export将您的 Python 依赖项导出到requirements.txt文件 - 使用
uv pip install为目标平台(aarch64-manylinux_2_28)安装依赖项
还会添加一个特定于您的 Agent 的 docker 目标,它将 Dockerfile 和打包的工件复制到 docker 上下文目录中。这将 Dockerfile 与构建输出放在一起,允许 CDK 使用 AgentRuntimeArtifact.fromAsset 直接构建 Docker 镜像。
为此项目构建的 Docker 镜像可以使用 Trivy 进行漏洞扫描,该工具从 ECR 托管的 Trivy 镜像运行。
项目中会添加一个 trivy 目标,用于扫描构建的镜像,如果发现任何 HIGH 或 CRITICAL 严重级别的漏洞,将以非零状态退出。生成的 Dockerfile 使用的基础镜像在生成时没有已知的可修复漏洞(这些严重级别),并升级捆绑的工具(如 npm)以保持这种状态。
扫描使用与镜像构建相同的容器引擎(docker 或 finch),因此不需要额外的工具。由于扫描仅在镜像更改时重新运行,未更改的镜像不会被重新扫描。提供的 trivy 根脚本会扫描工作区中的每个镜像:
pnpm trivyyarn trivynpm run trivybun trivy抑制 Trivy 发现
Section titled “抑制 Trivy 发现”在某些情况下,您可能希望抑制特定的漏洞,例如当尚无可用修复且您已评估风险为可接受时。
将漏洞 ID(每行一个)添加到项目根目录中的 .trivyignore 文件(即 project.json 旁边):
# node-tar arbitrary file write - not exploitable in our usageCVE-2024-XXXXX有关过滤发现的更多详细信息,请参阅 Trivy 过滤文档。
您的代理通过在 Dockerfile 中配置自动检测,使用 AWS Distro for Open Telemetry(ADOT)自动配置可观测性。
您可以在 CloudWatch AWS 控制台中找到跟踪,方法是在菜单中选择”GenAI Observability”。请注意,要填充跟踪,您需要启用事务搜索。
有关更多详细信息,请参阅 AgentCore 关于可观测性的文档。
session 选项根据您选择的框架映射到不同的底层持久性概念:Strands 的会话管理概念用于 strands 框架,或 LangGraph 的检查点概念用于 langchain 框架。
session 选项控制您的代理如何使用 Strands SDK 的 SessionManager 在调用之间持久化对话状态(消息历史、工具状态等):
s3(默认):CDK/Terraform 基础设施为会话数据提供专用的 S3 存储桶,使用专用的 KMS 密钥加密并阻止所有公共访问;服务器访问日志通过相同的密钥传送到 CloudWatch Logs 日志组。代理的 IAM 角色被授予对存储桶的读/写/列表/删除访问权限以及对密钥的解密/生成数据密钥访问权限,并且存储桶名称与代理的 ARN 一起注册在 AppConfig 运行时配置中。in-memory:不提供存储桶。对话状态仅在运行进程的生命周期内保留在内存中,不会在重启或缩容后保留。
这在生成的 session.py 中实现,它导出一个 get_session_manager() 函数,为当前会话解析 SessionManager。
由于 AG-UI 的 ag-ui-strands 适配器为每个对话线程克隆一个模板 Agent,因此 get_session_manager 作为 session_manager_provider 连接到传递给 main.py 中的 StrandsAgent 的 StrandsAgentConfig 上,因此为每个线程解析一个新的 SessionManager,而不是烘焙到共享模板代理中。
由于 with_session_id 已经为每个会话缓存一个 Agent 实例,因此 get_session_manager 直接在 agent.py 中的 get_agent 的 Agent(session_manager=get_session_manager()) 调用内调用。
会话 ID 本身来自 AgentCore Runtime 会话(通过 x-amzn-bedrock-agentcore-runtime-session-id 标头传播),并绑定到基于 contextvars.ContextVar 的上下文,因此 get_current_session_id() 可以在请求中的任何位置解析它 — 包括通过 connection 生成器连接的任何下游 MCP 或 A2A 客户端,因此整个调用链共享一致的会话。
session 选项控制您的代理的 LangGraph 检查点如何持久化对话状态:
s3(默认):部署的代理使用带有提供的会话存储桶的S3CheckpointSaver,将检查点和待处理写入存储在checkpoints/前缀下。此类位于共享代理连接项目中的s3_checkpoint_saver_langchain.py中。dynamodb-s3:CDK/Terraform 基础设施为检查点提供 DynamoDB 表,按照 AWS 关于使用 DynamoDB 作为 LangGraph 代理的检查点存储的文档中的建议进行配置(统一的PK/SK模式、PAY_PER_REQUEST计费、时间点恢复和ttl属性),以及用于卸载超过 350KB 的检查点的 S3 存储桶。两者都使用专用的 KMS 密钥加密;存储桶的服务器访问日志通过相同的密钥传送到 CloudWatch Logs 日志组。代理的 IAM 角色被授予对表和存储桶的读/写访问权限,并且表/存储桶名称与代理的 ARN 一起注册在 AppConfig 运行时配置中。in-memory:不提供表或存储桶。对话状态仅在运行进程的生命周期内保留在内存中,不会在重启或缩容后保留。
这在生成的 session.py 中实现,它导出一个从 agent.py 的 create_agent(..., checkpointer=get_checkpointer()) 调用的 get_checkpointer() 函数。
调用您的代理
Section titled “调用您的代理”调用本地服务器
Section titled “调用本地服务器”要通过 <your-agent-name>-serve 目标调用本地运行的 Agent,您可以向本地代理运行的端口上的 /invocations 发送一个简单的 POST 请求。例如,使用 curl:
curl -N -X POST http://localhost:8081/invocations \ -d '{"prompt": "what is 3 + 5?"}' \ -H "Content-Type: application/json"调用部署的代理
Section titled “调用部署的代理”要调用部署到 Bedrock AgentCore Runtime 的 Agent,您可以向 Bedrock AgentCore Runtime 数据平面端点发送 POST 请求,并使用 URL 编码的运行时 ARN。
您可以按以下方式从基础设施中获取运行时 ARN:
import { CfnOutput } from 'aws-cdk-lib';import { MyProjectAgent } from '@my-scope/common-constructs';
export class ExampleStack extends Stack { constructor(scope: Construct, id: string) { const agent = new MyProjectAgent(this, 'MyProjectAgent');
new CfnOutput(this, 'AgentArn', { value: agent.agentCoreRuntime.agentRuntimeArn, }); }}# Agentmodule "my_project_agent" { # Relative path to the generated module in the common/terraform project source = "../../common/terraform/src/app/agents/my-project-agent"
appconfig_application_id = module.runtime_config_appconfig.application_id appconfig_application_arn = module.runtime_config_appconfig.application_arn}
output "agent_arn" { value = module.my_project_agent.agent_core_runtime_arn}ARN 将具有以下格式:arn:aws:bedrock-agentcore:<region>:<account>:runtime/<agent-runtime-id>。
然后,您可以通过将 : 替换为 %3A 并将 / 替换为 %2F 来对 ARN 进行 URL 编码。
用于调用 Agent 的 Bedrock AgentCore Runtime 数据平面 URL 如下:
https://bedrock-agentcore.<region>.amazonaws.com/runtimes/<url-encoded-arn>/invocations调用此 URL 的确切方式取决于所使用的身份验证方法。
IAM 身份验证
Section titled “IAM 身份验证”对于 IAM 身份验证,请求必须使用 AWS Signature Version 4(SigV4)签名。
acurl <region> bedrock-agentcore -N -X POST \'https://bedrock-agentcore.<region>.amazonaws.com/runtimes/<url-encoded-arn>/invocations' \-d '{"prompt": "what is 3 + 5?"}' \-H 'Content-Type: application/json'Sigv4 enabled curl
您可以将以下脚本添加到您的 .bashrc 文件中(并 source 它),或者将以下内容粘贴到您希望运行命令的同一终端中。
acurl () { REGION=$1 SERVICE=$2 shift; shift; curl --aws-sigv4 "aws:amz:$REGION:$SERVICE" --user "$(aws configure get aws_access_key_id):$(aws configure get aws_secret_access_key)" -H "X-Amz-Security-Token: $(aws configure get aws_session_token)" "$@"}要进行 sigv4 认证的 curl 请求,请按如下方式调用 acurl:
acurl <region> <service> <other-curl-arguments>例如:
API Gateway
Section titled “API Gateway”acurl ap-southeast-2 execute-api -X GET https://xxxStreaming Lambda function url
Section titled “Streaming Lambda function url”acurl ap-southeast-2 lambda -N -X POST https://xxx您可以将以下函数添加到您的 PowerShell 配置文件中,或者将以下内容粘贴到您希望运行命令的同一 PowerShell 会话中。
# PowerShell profile or current sessionfunction acurl { param( [Parameter(Mandatory=$true)][string]$Region, [Parameter(Mandatory=$true)][string]$Service, [Parameter(ValueFromRemainingArguments=$true)][string[]]$CurlArgs )
$AccessKey = aws configure get aws_access_key_id $SecretKey = aws configure get aws_secret_access_key $SessionToken = aws configure get aws_session_token
& curl --aws-sigv4 "aws:amz:$Region`:$Service" --user "$AccessKey`:$SecretKey" -H "X-Amz-Security-Token: $SessionToken" @CurlArgs}要进行 sigv4 认证的 curl 请求,请使用以下示例调用 acurl:
API Gateway
Section titled “API Gateway”acurl ap-southeast-2 execute-api -X GET https://xxxStreaming Lambda function url
Section titled “Streaming Lambda function url”acurl ap-southeast-2 lambda -N -X POST https://xxxJWT / Cognito 身份验证
Section titled “JWT / Cognito 身份验证”对于 Cognito 身份验证,在 Authorization 标头中传递 Cognito 访问令牌:
curl -N -X POST 'https://bedrock-agentcore.<region>.amazonaws.com/runtimes/<url-encoded-arn>/invocations' \ -d '{"prompt": "what is 3 + 5?"}' \ -H "Content-Type: application/json" \ -H "Authorization: Bearer <access-token>"您可以使用 AWS CLI 的 cognito-idp admin-initiate-auth 命令获取访问令牌,例如:
aws cognito-idp admin-initiate-auth \ --user-pool-id <user-pool-id> \ --client-id <user-pool-client-id> \ --auth-flow ADMIN_NO_SRP_AUTH \ --auth-parameters USERNAME=<username>,PASSWORD=<password> \ --region <region> \ --query 'AuthenticationResult.AccessToken' \ --output text浏览器 / React 网站
Section titled “浏览器 / React 网站”要从 React 网站调用您的 Agent,您可以使用 connection 生成器,它会自动设置具有正确身份验证(IAM 或 Cognito)的客户端。
pnpm nx g @aws/nx-plugin:connectionyarn nx g @aws/nx-plugin:connectionnpx nx g @aws/nx-plugin:connectionbunx nx g @aws/nx-plugin:connection- 安装 Nx Console VSCode Plugin 如果您尚未安装
- 在VSCode中打开Nx控制台
- 点击
Generate (UI)在"Common Nx Commands"部分 - 搜索
@aws/nx-plugin - connection - 填写必需参数
- 点击
Generate
有关如何设置连接的详细信息,请参阅 connection 生成器指南。
将 A2A 代理作为工具调用
Section titled “将 A2A 代理作为工具调用”要将工作从此代理委托给远程 A2A 代理(TypeScript 或 Python),请使用 connection 生成器。它为目标代理提供 SigV4 身份验证的客户端,并对此代理的 agent.py 进行 AST 转换,以将远程 A2A 代理注册为 @tool 装饰的委托。
pnpm nx g @aws/nx-plugin:connectionyarn nx g @aws/nx-plugin:connectionnpx nx g @aws/nx-plugin:connectionbunx nx g @aws/nx-plugin:connection- 安装 Nx Console VSCode Plugin 如果您尚未安装
- 在VSCode中打开Nx控制台
- 点击
Generate (UI)在"Common Nx Commands"部分 - 搜索
@aws/nx-plugin - connection - 填写必需参数
- 点击
Generate
有关如何设置连接的详细信息,请参阅 connection 生成器指南。
调用 AG-UI 代理
Section titled “调用 AG-UI 代理”要从 React 网站调用您的 AG-UI 代理,请使用 connection 生成器,它为您部署的代理配置了具有正确身份验证(IAM 或 Cognito)的 CopilotKit 客户端。
pnpm nx g @aws/nx-plugin:connectionyarn nx g @aws/nx-plugin:connectionnpx nx g @aws/nx-plugin:connectionbunx nx g @aws/nx-plugin:connection- 安装 Nx Console VSCode Plugin 如果您尚未安装
- 在VSCode中打开Nx控制台
- 点击
Generate (UI)在"Common Nx Commands"部分 - 搜索
@aws/nx-plugin - connection - 填写必需参数
- 点击
Generate
有关如何设置连接的详细信息,请参阅 connection 生成器指南。
保护您的代理
Section titled “保护您的代理”Agent 会处理不受信任的输入,并可以通过其工具驱动实际操作,因此从一开始就考虑安全性是值得的。以下实践适用于生成的 agent。
将模型输入和输出视为不受信任
Section titled “将模型输入和输出视为不受信任”提示可能包含对抗性指令(提示注入),而模型输出是非确定性的——在安全敏感的逻辑中都不应信任它们:
- 为您的工具定义严格的输入模式,就像生成的示例工具中那样。将值限制为工具实际需要的内容(枚举、长度限制、数值范围),而不是接受自由格式的字符串。
- 切勿在没有验证或编码的情况下将模型输出直接传递到 shell 命令、SQL 查询、代码评估或渲染的 HTML 中。
- 在您的工具和下游服务中应用授权检查——不要依赖系统提示来防止模型滥用它有权访问的工具。
Strands 的 Prompt Engineering 和 Responsible AI 指南涵盖了编写健壮、具有安全意识的系统提示。
严格限制工具权限范围
Section titled “严格限制工具权限范围”仅授予 agent 的 IAM 角色其工具所需的权限。提供的 CDK 构造和 Terraform 模块公开了 grant* 方法和范围限定的策略以实现此目的——例如,授予 agent 访问权限以调用特定 API,而不是附加广泛的托管策略。当工具代表用户执行操作时,优先使用调用用户的身份(通过请求上下文传递)来授权操作,而不是使用 agent 自身的环境权限。
提供紧急开关
Section titled “提供紧急开关”由于模型行为可能以意外方式发生变化,因此请计划在不更改代码的情况下快速禁用或切换模型:
- 从配置中读取模型 ID(例如
MODEL_ID环境变量),以便操作员可以通过更新配置来切换或回滚到不同的模型。 - 在功能标志后面设置 agent,以便可以完全禁用其 AI 功能。禁用时,返回通用消息而不是错误,并确保应用程序的其余部分能够优雅降级。
在您的操作手册中记录如何切换这些控制。
保护敏感数据
Section titled “保护敏感数据”- 避免记录提示和完成内容,它们可能包含用户数据。生成的 agent 的模型错误日志记录钩子仅记录错误元数据,而不记录对话内容——在添加自己的日志记录时保持此属性。
- 向用户返回通用错误消息;在服务器端记录详细错误。
- 在用户和会话之间隔离对话状态,并授权访问任何持久化的会话数据。
- 从提示和输出中编辑个人身份信息(PII)——可以使用 Bedrock Guardrail 敏感信息过滤器(见下文),或者对于 Strands agent,使用 PII Redaction 指南中的方法。
Amazon Bedrock Guardrails
Section titled “Amazon Bedrock Guardrails”Amazon Bedrock Guardrails 提供可配置的内容过滤器、拒绝主题和敏感信息(PII)过滤器,这些过滤器会在模型输入和输出时进行评估。您可以将护栏附加到生成的 agent 使用的模型:
import os
from strands import Agentfrom strands.models import BedrockModel
model = BedrockModel( model_id=os.environ.get("MODEL_ID"), guardrail_id=os.environ["GUARDRAIL_ID"], guardrail_version=os.environ.get("GUARDRAIL_VERSION", "DRAFT"),)
agent = Agent(model=model)有关更多详细信息,请参阅 Strands Guardrails 指南。
import os
from langchain_aws import ChatBedrockConverse
model = ChatBedrockConverse( model=os.environ.get("MODEL_ID"), guardrail_config={ "guardrailIdentifier": os.environ["GUARDRAIL_ID"], "guardrailVersion": os.environ.get("GUARDRAIL_VERSION", "DRAFT"), },)有关 guardrail_config 字段,请参阅 ChatBedrockConverse 文档。
使用 connection 生成器将此项目与工作区中的其他项目集成。以下连接涉及此项目:
