使用 Amazon Bedrock AgentCore 构建主权风险交易代理，分析收益率差、FX 对冲与债务重新定价
Amazon Bedrock AgentCore 协助团队构建生产等级的主权风险交易代理，分析政府国债收益率差、信用价差扩大、债务重新定价、FX 波动、流动性压力与防御性资产轮动。本指南逐步说明 Runtime 部署、Gateway、MCP 工具服务器、Cognito 验证、流式响应、会话管理与大型负载处理，以便在企业交易工作流中提供安全、受治理的实时市场情报。
金融服務系列
- 第 1 篇：主權風險交易代理（殖利率差、FX、債務重新定價）
- 第 2 篇：現代波動率交易與合法泰國復原規劃多代理系統
- 第 3 篇：空頭跨式交易風險治理與 AgentCore Memory
- 第 4 篇：在 Amazon EKS 上建構生產就緒信用與收益質押 AI 智能體
免责声明
并非投资建议
本文提供的信息仅供一般教育与风险意识用途，不构成财务、投资、税务、法律或交易建议，也不建议买入、卖出或持有任何金融工具。
仅限合法的泰国规划
本助理仅为合法旅客提供符合法律且透明的泰国旅游、搬迁、度假与休养规划决策支持，包括行程构想、交通与后勤、文化准备、住宿研究及健康照护选项。
不提供非法规避或逃亡协助
在任何情况下，本服务均不会协助藏匿、逃避主管机关、规避移民法规、法院命令、债务、制裁或法律义务，也不会协助非法逃亡、制作伪造文件、隐瞒行踪或进行任何非法脱逃。
卖出跨式交易的风险治理范围
卖出跨式交易风险治理内容遵循上传的架构，着重 Delta、Gamma、Vega、保证金、流动性、压力损失、反马丁格尔行为、限额、升级处理、监控、文件记录、批准与治理控制，以保持严谨的监督纪律。
演示：主权风险 AgentCore Runtime 质量
用户：
分析美中政府国债收益率差，以决定下一个头寸行动。评估久期需求、中国成长风险、资本流动、RMB 压力、UST-CGB 相对价值、信用价差扩大与主权债务重新定价。
AgentCore Runtime 输出：
先判断风险环境，而非交易方向。美中收益率差扩大可能代表市场对美国长久期债券的需求增强、对中国成长的信心减弱、资本流动承压、RMB 贬值风险升高，或美国国债与中国政府国债之间的相对价值重估。下一步应以受治理的风险审查为框架：降低风险、防御性增加久期、对冲货币风险敞口、轮动至较高质量的主权债券，或等待确认。确认信号包括 Fed 政策预期、PBOC 宽松政策、中国信用脉冲、USD/CNH 走势、实际收益率变化、外资流入或流出、流动性与风险偏好。
演示：Gateway 工具选择质量
用户：
找出适合分析主权债务重新定价、信用价差扩大与 FX 压力的工具。
AgentCore Gateway 输出：
先使用语义工具探索，而非将所有结构描述载入提示。相关工具可能包括 sovereign_debt_risk_repricing、credit_spread_widening、funding_liquidity、currency_mismatch、policy_re行动_function 与 correlation_shock。Gateway 应强制执行验证，将 MCP 呼叫转译至目标 Lambda 或 API，并保留延迟、呼叫状态与错误分类的可观测性。
目录
第 1 部分：执行概述
推出生产级主权风险交易代理，使用 AgentCore Runtime、网关、MCP 工具、身份验证、流式、会话和负载处理，以实现安全的全球机构市场情报和大规模托管工作流。
第 2 部分：代理即工具架构
解释了一个分层编排模型，其中中央宏观代理动态地将生存、周期性和冲击资本任务委托给涉及美元流动性、中国需求和欧洲信用风险制度的专门子代理。
第 3 部分：构建运行时代理
展示如何使用 Strands、Amazon Nova Pro、计算器工具以及生产环境中主权信用和收益率利差分析的系统提示来安全部署无服务器 AgentCore Runtime 应用。
第 4 部分. 呼叫运行时代理
当演示通过 boto3 呼叫已部署的运行时，传递 JSON 提示、会话标识符和负载字节以返回模型响应，从而实现跨桌面应用和服务的安全实时交易工作流整合。
第 5 部分. 构建 MCP 工具服务器
定义无状态可流式 HTTP MCP 服务器，该服务器安全地公开特定领域的工具，用于流动性、信用利差、主权重新定价、FX 波动性、商品冲击、CDS 对冲和 AgentCore Runtime 部署内的相关风险。
第 6 部分. 本地和远程 MCP 测试
提供本地和远程 MCP 客户端模式，用于工具探索、会话初始化、Cognito 持有者令牌身份验证以及生产部署之前的运行时端点测试以及跨受控环境的更广泛代理整合。
第 7 部分. 配置 Amazon Cognito 身份验证
描述 Cognito 用户池、自定义 JWT 授权者、允许的客户端、发现 URL 和刷新令牌逻辑，用于通过企业身份控制和可审核性要求保护入站 AgentCore Runtime 和 MCP 网关访问。
第 8 部分：流式 AgentCore Runtime 响应
解释异步入口点、Server-Sent Events、分块输出、内容类型处理和 boto3 流式处理，以在波动的交易和风险监控会话期间提供安全、低延迟的市场评论和响应迅速的代理体验。
第 9 部分. 管理 AgentCore Runtime 会话
涵盖运行时会话生命周期、对话历史记录、应用状态、会话标识符、不活动终止以及用于场景多轮分析的明确停止呼叫以及跨机构工作流的受控资源使用情况。
第 10 部分：处理大型负载
展示如何在最大负载限制下安全地处理 Base64 编码的 Excel 文件和图像，将结构化数据与图表信号相结合以进行多模态财务分析和投资组合风险审查工作流。
第 11 部分：Amazon Bedrock AgentCore网关
引入 AgentCore Gateway 作为托管 MCP 整合层，将 API、Lambda、Smithy 和 OpenAPI 定义转换为安全、代理可访问的企业工具，具有身份验证内置功能、可观测性和治理控制功能。
第 12 部分：将 Lambda 函数转换为MCP工具
概述如何构建 Lambda 支持的 MCP 工具、IAM 角色、Cognito 授权、网关目标、工具架构和凭证提供程序，以便通过代理在安全的 AWS 生产环境和工作流中实现一致、受控的业务逻辑。
第 13 部分. 从 Strands Agent 呼叫 AgentCore Gateway
演示如何将 Strands Agent 与经过 SigV4 验证的 MCP 传输、临时 STS 凭证、工具探索、直接工具呼叫以及通过 AgentCore Gateway 进行代理驱动的执行结合使用，以实现安全的企业整合和受控自动化场景。
第 14 部分.AgentCore Gateway 语义搜索
解释 Gateway 如何通过嵌入对工具元数据进行索引并公开语义搜索，以便智能体发现相关工具，而无需将每个模式加载到提示中，从而显著降低延迟成本和复杂性。
第 15 部分：主权风险交易方法
围绕着成长、流动性、偿付能力、货币实力、政策反应函数、资本流动、信用利差迁移、FX 压力以及机构主权风险决策和对冲的跨资产轮换构建市场方法论。
第 16 部分：历史危机背景
使用长期萧条、霸菱危机、经济大恐慌、石油冲击、亚洲金融风暴和 LTCM 等历史危机来解释流动性重新定价、杠杆传染和防御性投资组合轮换纪律教训。
第 1 部分：执行概述
推出生产级主权风险交易代理，使用 AgentCore Runtime、网关、MCP 工具、身份验证、流式、会话和负载处理，以实现安全的全球机构市场情报和大规模托管工作流。
Amazon Bedrock AgentCore 可用于为机构金融服务工作流构建安全的生产级主权风险交易代理。
代理可以分析：
- 政府国债收益率利差
- 信用利差扩大
- 主权债务重新定价
- FX 波动性
- 资金流动性
- 资本流动
- 投资组合曝光
- 防御性资产轮换
- 政策反应函数
- 跨资产风险迁移
该架构结合了：
- AgentCore Runtime 用于可扩充代理执行
- AgentCore Gateway 用于工具探索和 API 转换
- 用于结构化金融工具的MCP服务器
- Amazon Cognito 用于认证
- 实时市场评论的流式响应
- 多轮工作流的会话管理
- Excel 文件、图表和多模态数据的大负载处理
第 2 部分：代理即工具架构
解释了一个分层编排模型，其中中央宏观代理动态地将生存、周期性和冲击资本任务委托给涉及美元流动性、中国需求和欧洲信用风险制度的专门子代理。
智能体作为工具
描述
智能体使用分层架构。中央协调代理管理严格的三层资本策略：
- 生存资本
- 景气循环资本
- 冲击资本
协调者不断评估宏观经济变化，并将专门的资产配置任务委托给子代理，重点关注：
- 美元流动性周期
- 中国工业需求
- 欧洲主权信用断裂
- 全球风险溢酬移民
- 多资产防守轮换
益处
这种解耦设计降低了市场非理性繁荣时期发生灾难性损失的风险。
通过将宏观策略监管与本地交易执行分开，金融机构可以在市场共识完全形成之前执行有纪律的资本迁移。这有助于减少接触：
- 影子银行杠杆
- 流动性幻想
- 交易拥挤
- 相关性冲击
- 突然的系统性重新定价
- 隐藏的全球风险溢价冲击
使用案例
投资银行可以使用这个协调器来监控政策反应功能。
例如：
- 如果中国基础设施订单增加，协调者可以委托周期性代理来分析铜、航运股票、资源股票和高贝塔亚洲股票。
- 如果欧洲主权银行利差扩大，协调者可以触发冲击资本代理来分析黄金、波动性工具、信用违约掉期对冲和美元流动性头寸。
演示
AgentCore Runtime输入：
您是资深机构市场专家，专门研究信用利差扩大、主权债务风险重新定价和跨资产风险轮换。您可以执行数学计算。提供有关主权风险、中美国债收益率利差、信用利差、久期、FX 对冲、流动性、资本流向、投资组合风险敞口和防御性定位的简明交易观点。当提供 Excel 文件或图像时，结合结构化数据和视觉信号来提供见解、交易影响、对冲想法、确认信号和失效触发器。
AgentCore Runtime输出：
信贷的核心关键字是成长、流动性、偿付能力、金钱实力和政治动员能力。美国拥有 Federal 储备、储备货币地位、深厚的资本市场和量化宽松政策。因此，美国的信贷危机可以通过央行资产负债表的扩张来重新定价。
中国受惠于WTO红利、国营银行、基础建设需求、资本管制和政策动员能力。因此，中国信贷市场并非纯粹以市场为基础的违约风险定价，而是作为政策反应函数定价。中国的风险在于对大宗商品的依赖、地方政府债务、产能过剩和环境压力。
欧洲拥有强势货币，但缺乏统一的财政权力。因此，欧元区的弱点并非货币本身，而是主权信用与银行体系之间的周期性压力。
信用风险关键字包括资金流动性、信用利差迁移、货币错配、政策反应函数、拥挤交易和相关性冲击。中国信用交易尤其要关注「国十措施」、「新国十措施」和中期减债。
当保单扩张时，信用利差就会被压缩。当政策聚焦稳定成长时，国营企业、地方政府融资平台、基础建设链都得到支持。当政策转向去杠杆时，弱的民企、弱的房企、弱势的地方平台都会重新定价。
卡内基对中国金融体系的研究表明，中国的金融体系既可以服务成长，又可以累积下行风险。
接下来的交易动作应该是基于三层轮替结构：生存资本、景气循环资本、冲击资本。
生存资本用于确保不被市场强迫清算。应该配置为美元现金、短期美国国债、高等级信用债券和黄金。
景气循环资本被用来跟随主导的宏观引擎。如果美国企业获利、就业、房地产和信用利差改善，应该购买美国股票、资本重组的金融股、投资等级信用债券和有实际现金流的科技股。
如果中国的PMI、铜价、煤炭价格、铁矿石、亚洲出口和基础设施订单上涨，则应该买入中国相关的周期股、资源股、航运股和高贝塔值亚洲股票。
如果欧洲因PIIGS利差扩大、银行业压力和欧元贬值而走弱，人们应该减少欧元信贷风险敞口，避开疲软的银行，并持有德国国债或美元资产。
冲击资本用于购买保护：黄金、波动性工具、信用违约掉期对冲、美元多头头寸和现金储备。
戴尔·阿里西亚和毛罗的《重新思考宏观经济政策》的交易教训是，真正的赢家不是预测每一次危机的人，而是能够在宏观政策变化时执行轮换纪律的人。
卡内基对资本流动风险和全球再平衡的研究也表明，宏观风险会在股票、债券、外汇、大宗商品和信贷之间转移，而不是停留在单一市场。
1873年至1879年，铁路、银行、产能、跨国信贷共同造成了长期萧条，产生了长期的通货紧缩调整。与2008年类似，过度融资满足了需求不足，杠杆乐观最终转化为资产负债表之痛。结果是清算、低成长和反全球化压力。
1890年的霸菱危机见证了阿根廷债务压力冲击伦敦金融体系，显示外围市场的信用风险可能威胁到核心金融中心。与欧元区债务危机和新兴市场资本外流类似，信心取决于债权人的协调。结果是救援工作阻止了崩溃，但风险被重新定价。
从 1929 年到 1933 年，经济大恐慌见证了股市狂热、债务、脆弱的银行和政策错误，导致市场衰退演变成经济崩溃。与2008年类似，杠杆和资产泡沫感染了实体经济。结果是去杠杆化、失业和国家强力干预。
从 1973 年到 1975 年，石油冲击通过大宗商品价格冲击、通货膨胀、低成长和政策混乱对传统投资组合造成了冲击。与中国大宗商品超级周期逆转的风险类似，能源和原物料改变了宏观定价。结果是资本转向抗通膨资产。
1997年至1998年，亚洲金融风暴和LTCM事件涉及资本流向逆转、货币崩盘和杠杆放大压力。与卡内基对危险资本流动的警告类似，吸收不稳定资本的经济体是最脆弱的。结果是蔓延、被迫抛售和流动性重新定价。
Amazon Bedrock AgentCore 核心元件
Amazon Bedrock AgentCore Runtime（文章介绍）
用于部署可扩展代理的无服务器执行时间。优点：安全隔离和快速启动。使用案例：跨框架和模型的生产编排。
Amazon Bedrock AgentCore Gateway（本文涵盖）
代理的工具探索和 API 转换。优点：更快的整合和更低的延迟。用例：代理根据场景选择相关工具。
Amazon Bedrock AgentCore Identity
代理身份管理整合提供者。好处：授权访问，减少同意疲劳。使用案例：代理访问 AWS 和第三方工具。
Amazon Bedrock AgentCore Code Interpreter
代理的沙盒代码执行。优点：通过可配置的隔离确保分析安全。使用案例：运行安全计算、转换和数据工作流。
Amazon Bedrock AgentCore Browser
用于网站互动的云浏览器运行时。好处：安全的扩展和可观测性。使用案例：自动化网络研究、表单填写和测试。
Amazon Bedrock AgentCore Memory
场景感知代理的托管内存。好处：控制短期和长期回忆。使用案例：安全地跨会话和代理进行个人化对话。
Amazon Bedrock AgentCore Observability
监控生产中的代理工作流。优点：通过遥测仪表板进行更快的侦错。使用案例：跟踪延迟、故障和工具行为。
Amazon Bedrock AgentCore Evaluations
代理质量评估管理。好处：对正确性、安全性和有用性进行持续评分。用例：使用内置或自定义评估器进行生产治理。
Amazon Bedrock AgentCore Policy
用于验证工具呼叫的策略层。好处：细致的动作控制。使用案例：安全地阻止未经授权的数据、系统或工具访问。
第 3 部分：构建运行时代理
展示如何使用 Strands、Amazon Nova Pro、计算器工具以及生产环境中主权信用和收益率利差分析的系统提示来安全部署无服务器 AgentCore Runtime 应用。
AgentCore Runtime既支持传统的HTTP请求响应模式，也支持基于MCP的工具协议。
- HTTP 协议：传统 REST 风格的请求和响应端点
- MCP协议：Model Context Protocol用于工具和代理服务器
- 预设连接埠：代理托管在连接埠 8080 上
from strands import Agent
from strands_tools import calculator
from bedrock_agentcore.runtime import BedrockAgentCore应用
from strands.models import BedrockModel
应用 = BedrockAgentCore应用()
model = BedrockModel(
model_id="amazon.nova-pro-v1:0"
)
agent = Agent(
model=model,
tools=[calculator],
system_提示词="You're a credit spread widening and sovereign debt risk repricing expert. You can do math calculations."
)
@应用.entrypoint
def strands_agent_bedrock(payload):
user_input = payload.get("提示词")
response = agent(user_input)
return response.message["content"][0]["text"]
if __name__ == "__main__":
应用.run()
第 4 部分：呼叫运行时代理
当演示通过 boto3 呼叫已部署的运行时，传递 JSON 提示、会话标识符和负载字节以返回模型响应，从而实现跨桌面应用和服务的安全实时交易工作流整合。
import boto3, json
client = boto3.client("bedrock-agentcore", region_name="us-east-1")
response = client.invoke_agent_runtime(
agentRuntimeArn=agent_arn,
runtimeSessionId=session_id,
payload=json.dumps({"提示词": 提示词}).encode()
)
第 5 部分. 构建 MCP 工具服务器
定义无状态可流式 HTTP MCP 服务器，该服务器安全地公开特定领域的工具，用于流动性、信用利差、主权重新定价、FX 波动性、商品冲击、CDS 对冲和 AgentCore Runtime 部署内的相关风险。
MCP 服务器使用无状态流式 HTTP 和 Mcp-Session-Id 标头进行会话隔离。
主要特点：
- 运行于 0.0.0.0:8000
- 公开 /mcp 端点
- 在AgentCore Runtime内部运行
- 提供用于主权风险和跨资产分析的金融领域工具
具体工具：
- usd_liquidity_cycle()
- china_industrial_demand_engine()
- european_sovereign_credit_fractures()
- global_risk_premium_migration()
-纪律多元资产轮替（） - Credit_spread_widening()
- 银行融资压力()
- 主权债务风险重新定价（）
- heightened_fx_volatility()
- Sharp_commodity_price_volatility()
- Credit_default_swap_hedging()
-funding_liquidity() - Credit_spread_migration()
- 货币_不符()
- 策略反应函数()
- 拥挤的交易()
-correlation_shock()
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(host="0.0.0.0", stateless_http=True)
@mcp.tool()
def usd_liquidity_cycle() -> str:
"""Analyze the USD liquidity cycle"""
return "USD liquidity cycle analysis"
@mcp.tool()
def china_industrial_demand_engine() -> str:
"""Analyze China's industrial demand engine"""
return "China industrial demand engine analysis"
@mcp.tool()
def european_sovereign_credit_fractures() -> str:
"""Analyze European sovereign credit fractures"""
return "European sovereign credit fractures analysis"
@mcp.tool()
def global_risk_premium_migration() -> str:
"""Analyze global risk premium migration"""
return "Global risk premium migration analysis"
@mcp.tool()
def disciplined_multi_asset_rotation() -> str:
"""Analyze disciplined multi-asset rotation"""
return "Disciplined multi-asset rotation analysis"
@mcp.tool()
def credit_spread_widening() -> str:
"""Analyze credit spread widening"""
return "Credit spread widening analysis"
@mcp.tool()
def bank_funding_pressure() -> str:
"""Analyze bank funding pressure"""
return "Bank funding pressure analysis"
@mcp.tool()
def sovereign_debt_risk_repricing() -> str:
"""Analyze sovereign debt risk repricing"""
return "Sovereign debt risk repricing analysis"
@mcp.tool()
def heightened_fx_volatility() -> str:
"""Analyze heightened foreign exchange volatility"""
return "Heightened FX volatility analysis"
@mcp.tool()
def sharp_commodity_price_volatility() -> str:
"""Analyze sharp commodity price volatility"""
return "Sharp commodity price volatility analysis"
@mcp.tool()
def credit_default_swap_hedging() -> str:
"""Analyze credit default swap hedging"""
return "Credit default swap hedging analysis"
@mcp.tool()
def funding_liquidity() -> str:
"""Analyze funding liquidity"""
return "Funding liquidity analysis"
@mcp.tool()
def credit_spread_migration() -> str:
"""Analyze credit spread migration"""
return "Credit spread migration analysis"
@mcp.tool()
def currency_mismatch() -> str:
"""Analyze currency mismatch"""
return "Currency mismatch analysis"
@mcp.tool()
def policy_re行动_function() -> str:
"""Analyze policy re行动 function"""
return "Policy re行动 function analysis"
@mcp.tool()
def crowded_trades() -> str:
"""Analyze crowded trades"""
return "Crowded trades analysis"
@mcp.tool()
def correlation_shock() -> str:
"""Analyze correlation shock"""
return "Correlation shock analysis"
if __name__ == "__main__":
mcp.run(transport="streamable-http")
第 6 部分. 本地和远程 MCP 测试
提供本地和远程 MCP 客户端模式，用于工具探索、会话初始化、Cognito 持有者令牌身份验证以及生产部署之前的运行时端点测试以及跨受控环境的更广泛代理整合。
构建并运行本地测试客户端
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
async def main():
async with streamablehttp_client(
"http://localhost:8000/mcp", {}, timeout=120, terminate_on_close=False
) as (read, write, _):
async with ClientSession(read, write) as session:
await session.initialize()
return await session.list_tools()
if __name__ == "__main__":
asyncio.run(main())
终端机1：启动MCP服务器
python mcp_server.py
终端机 2：运行测试客户端
python mcp_client.py
构建远程测试客户端
远程客户端撷取执行时间 ARN，根据需要刷新 Cognito 令牌，并连接到 AgentCore Runtime MCP 端点。
import asyncio, base64, json, time, boto3
from datetime import timedelta
from urllib.parse import quote
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
def refresh_token(client_id, refresh_token, region):
return boto3.client("cognito-idp", region_name=region).initiate_auth(
ClientId=client_id,
AuthFlow="REFRESH_TOKEN_AUTH",
AuthParameters={"REFRESH_TOKEN": refresh_token},
)["AuthenticationResult"]["AccessToken"]
def valid_token(token, client_id, refresh, region):
try:
payload = token.split(".")[1] + "=" * (-len(token.split(".")[1]) % 4)
exp = json.loads(base64.b64decode(payload))["exp"]
return refresh_token(client_id, refresh, region) if exp - time.time() < 300 else token
except Exception:
return refresh_token(client_id, refresh, region)
async def main():
region = boto3.session.Session().region_name
agent_arn = boto3.client("ssm", region_name=region).get_parameter(
Name="/mcp_server/runtime/agent_arn"
)["Parameter"]["Value"]
token = valid_token(
cognito_config["bearer_token"],
cognito_config["client_id"],
cognito_config["refresh_token"],
region,
)
url = (
f"https://bedrock-agentcore.{region}.amazonaws.com/runtimes/"
f"{quote(agent_arn, safe='')}/invocations?qualifier=DEFAULT"
)
async with streamablehttp_client(
url,
{"authorization": f"Bearer {token}", "Content-Type": "应用lication/json"},
timeout=timedelta(seconds=120),
terminate_on_close=False,
) as (read, write, _):
async with ClientSession(read, write) as session:
await session.initialize()
return await session.list_tools()
if __name__ == "__main__":
asyncio.run(main())
远程运行MCP客户端
python mcp_server_remote.py
第 7 部分. 配置 Amazon Cognito 验证
描述 Cognito 用户池、自定义 JWT 授权者、允许的客户端、发现 URL 和刷新令牌逻辑，用于通过企业身份控制和可审核性要求保护入站 AgentCore Runtime 和 MCP 网关访问。
创建Amazon Cognito进行认证
import sys
sys.path.insert(0, "../..")
from utils import setup_cognito_user_pool
cognito_config = setup_cognito_user_pool()
配置AgentCore Runtime
import os
from bedrock_agentcore_starter_toolkit import Runtime
auth_config = {
"customJWTAuthorizer": {
"allowedClients": [cognito_config["client_id"]],
"discoveryUrl": cognito_config["discovery_url"],
}
}
response = Runtime().configure(
entrypoint="mcp_server.py",
auto_create_execution_role=True,
auto_create_ecr=True,
requirements_file="requirements.txt",
region=region,
authorizer_configuration=auth_config,
protocol="MCP",
agent_name=tool_name,
)
推出AgentCore Runtime
launch_result = agentcore_runtime.launch()
输出：agent_arn、agent_id
第 8 部分：流式 AgentCore Runtime 响应
解释异步入口点、Server-Sent Events、分块输出、内容类型处理和 boto3 流式处理，以在波动的交易和风险监控会话期间提供安全、低延迟的市场评论和响应迅速的代理体验。
AgentCore Runtime 流式响应
AgentCore Runtime 支持通过异步入口点进行流式响应。
关键概念：
- 对入口点函数使用 async def
- 使用yield来传输块
- 返回Server-Sent Events格式
- 使用内容类型：文字/事件流
- 当函数完成或引发未处理的异常流结束
from strands import Agent, tool
from bedrock_agentcore.runtime import BedrockAgentCore应用
from strands.models import BedrockModel
应用 = BedrockAgentCore应用()
@tool
def credit_spread_widening():
"""Analyze credit spread widening"""
return "Credit spread widening analysis"
@tool
def sovereign_debt_risk_repricing():
"""Analyze sovereign debt risk repricing"""
return "Sovereign debt risk repricing analysis"
agent = Agent(
model=BedrockModel(model_id="amazon.nova-pro-v1:0"),
tools=[credit_spread_widening, sovereign_debt_risk_repricing],
system_提示词="You're a credit spread widening and sovereign debt risk repricing expert. You can do math calculations." ,
)
@应用.entrypoint
async def strands_agent_bedrock_streaming(payload):
try:
async for event in agent.stream_async(payload.get("提示词")):
if "data" in event:
yield event["data"]
except Exception as e:
yield {"error": str(e), "type": "stream_error"}
if __name__ == "__main__":
应用.run()
部署 AgentCore Runtime 流式响应
将代码打包为容器，并使用 CI/CD 管道和 IaC 推送到 ECR。
使用 Amazon Bedrock AgentCore Python SDK 打包工件并部署到 AgentCore Runtime。
from boto3.session import Session
from bedrock_agentcore_starter_toolkit import Runtime
response = Runtime().configure(
entrypoint="strands_claude_streaming.py",
auto_create_execution_role=True,
auto_create_ecr=True,
requirements_file="requirements.txt",
region=Session().region_name,
agent_name="strands_claude_streaming",
)
启动 AgentCore Runtime 流式响应
launch_result = agentcore_runtime.launch()
通过流直接呼叫 AgentCore Runtime
invoke_response = agentcore_runtime.invoke({
"提示词": "Focus specifically on the US-China government bond yield spread and what it implies for the next position 行动. Assess whether the spread movement suggests stronger demand for US duration, weaker confidence in China growth, capital flow pressure, currency pressure on the renminbi, or a relative-value opportunity between US Treasuries and Chinese government bonds. Provide a practical trading view rather than a theoretical essay. Explain whether the preferred 行动 is to reduce risk, add duration, hedge currency exposure, rotate into higher-quality sovereign bonds, stay defensive, or wait for confirmation. Include the key signals to watch: Fed policy expectations, PBOC easing, China credit demand, USD/CNH movement, real yield changes, foreign inflows or outflows, and risk 应用etite. Finally, give clear invalidation points that would require changing the position quickly."
})
invoke_response
使用 AgentCore Runtime 流式响应
处理 Server-Sent Events 格式。
使用boto3进行流式
import boto3, json
提示词 = "Analyze the US-China government bond yield spread for next-position 行动. Assess duration demand, China growth risk, capital flows, RMB pressure, and UST-CGB relative value. Provide a professional trading view covering risk bias, duration positioning, FX hedging, sovereign rotation, confirmation signals, and invalidation triggers. Key factors: Fed path, PBOC easing, China credit impulse, USD/CNH, real yields, cross-border flows, liquidity, and risk 应用etite."
res = boto3.client("bedrock-agentcore", region_name=region).invoke_agent_runtime(
agentRuntimeArn=launch_result.agent_arn,
qualifier="DEFAULT",
payload=json.dumps({"提示词": 提示词}),
)
if "text/event-stream" in res.get("contentType", ""):
response = " ".join(
line.decode("utf-8")[6:].replace('"', "")
for line in res["response"].iter_lines(chunk_size=1)
if line and line.decode("utf-8").startswith("data: ")
)
else:
events = list(res.get("response", []))
response = json.loads(events[0].decode("utf-8")) if events else None
response
流式响应 Markdown AgentCore Runtime
import boto3, json
from IPython.display import Markdown, display
提示词 = "Analyze the US-China government bond yield spread for next-position 行动. Assess duration demand, China growth risk, capital flows, RMB pressure, and UST-CGB relative value. Provide a professional trading view covering risk bias, duration positioning, FX hedging, sovereign rotation, confirmation signals, and invalidation triggers. Key factors: Fed path, PBOC easing, China credit impulse, USD/CNH, real yields, cross-border flows, liquidity, and risk 应用etite."
res = boto3.client("bedrock-agentcore", region_name=region).invoke_agent_runtime(
agentRuntimeArn=launch_result.agent_arn,
qualifier="DEFAULT",
payload=json.dumps({"提示词": 提示词}),
)
if "text/event-stream" in res.get("contentType", ""):
response = " ".join(
line.decode("utf-8")[6:].replace('"', "")
for line in res["response"].iter_lines(chunk_size=1)
if line and line.decode("utf-8").startswith("data: ")
)
else:
events = list(res.get("response", []))
response = json.loads(events[0].decode("utf-8")) if events else ""
display(Markdown(response))
第 9 部分. 管理 AgentCore Runtime 会话
涵盖运行时会话生命周期、对话历史记录、应用状态、会话标识符、不活动终止以及用于场景多轮分析的明确停止呼叫以及跨机构工作流的受控资源使用情况。
AgentCore Runtime 会话管理
AgentCore Runtime 会话支持场景、多轮工作流。
会话生命周期
- 构建：第一次呼叫构建会话。
- 活动状态：会话处理请求。
- 空闲状态：会话等待下一次呼叫。
- 终止：会话在不活动或最大生命周期限制后终止。
会话状态
会话可以保留：
- 对话历史记录
- 之前的回复
- 应用状态
- 运行时变量
- 会话期间构建或修改的文件
from strands import Agent, tool
from strands_tools import calculator
from bedrock_agentcore.runtime import BedrockAgentCore应用
from strands.models import BedrockModel
应用 = BedrockAgentCore应用()
@tool
def credit_spread_widening():
"""Assess credit spread widening, funding stress, liquidity risk, and risk-asset pressure."""
return "Credit spread widening may signal rising default risk, tighter funding, weaker liquidity, and defensive risk positioning."
@tool
def sovereign_debt_risk_repricing():
"""Assess sovereign debt repricing, yield spread pressure, duration risk, and FX implications."""
return "Sovereign debt repricing may reflect policy divergence, fiscal risk, real-yield shifts, capital flows, and currency pressure."
agent = Agent(
model=BedrockModel(model_id="amazon.nova-pro-v1:0"),
tools=[calculator, credit_spread_widening, sovereign_debt_risk_repricing],
system_提示词= “You're a professional FSI markets assistant. Always acknowledge the user's name. Provide concise institutional trading views on credit spreads, sovereign risk, US-China bond yield spread, duration, FX hedging, liquidity, and risk rotation.",
)
users = {
"1": "María García",
"2": "Carlos Rodríguez",
"3": "Lucía Fernández",
"4": "Diego Martínez",
"5": "Sofía López",
}
@应用.entrypoint
def strands_agent_bedrock_handling_context(payload, context):
name = users[payload["user_id"]]
提示词 = f"My name is {name}. Request: {payload.get('提示词')}\nSession: {context.session_id}. Analyze the US-China government bond yield spread for next-position 行动. Assess duration demand, China growth risk, capital flows, RMB pressure, UST-CGB relative value, credit spread widening, and sovereign debt repricing. Provide risk bias, trade 行动, hedge options, confirmation signals, and invalidation triggers. Please acknowledge my name and assist."
return agent(提示词).message["content"][0]["text"]
if __name__ == "__main__":
应用.run()
构建AgentCore Runtime会话管理的会话
import uuid
from IPython.display import Markdown, display
session_id = str(uuid.uuid4())
invoke_response = agentcore_runtime.invoke(
{
"提示词": "Analyze the US-China government bond yield spread for next-position 行动. Assess duration demand, China growth risk, capital flows, RMB pressure, and UST-CGB relative value. Provide a professional trading view covering risk bias, duration positioning, FX hedging, sovereign rotation, confirmation signals, and invalidation triggers. Key factors: Fed path, PBOC easing, China credit impulse, USD/CNH, real yields, cross-border flows, liquidity, and risk 应用etite.",
"user_id": "1",
},
session_id=session_id,
)
display(Markdown(invoke_response["response"][0]))
停止AgentCore Runtime会话管理的会话
import boto3
boto3.client("bedrock-agentcore", region_name=region).stop_runtime_session(
agentRuntimeArn=launch_result.agent_arn,
runtimeSessionId=str(session_id),
qualifier="DEFAULT",
)
第 10 部分：处理大型负载
展示如何在最大负载限制下安全地处理 Base64 编码的 Excel 文件和图像，将结构化数据与图表信号相结合以进行多模态财务分析和投资组合风险审查工作流。
AgentCore Runtime 大负载搬运
AgentCore Runtime可以处理大型负载，例如Excel文件和镜像。这对于涉及以下内容的金融服务工作流非常有用：
- 债券收益率数据集
- 信用利差数据集
- 主权风险仪表板
- FX图表
- 流动性报告
- 投资组合曝光文件
- 资本流分析
重要的：
负载高达 100MB
import base64
from strands import Agent
from strands.models import BedrockModel
from bedrock_agentcore.runtime import BedrockAgentCore应用
应用 = BedrockAgentCore应用()
agent = Agent(
model=BedrockModel(
model_id="amazon.nova-pro-v1:0",
max_tokens=16000,
),
system_提示词="You are a professional FSI data analysis assistant. Analyze Excel files containing financial services data such as bond yields, credit spreads, sovereign risk, FX, liquidity, capital flows, and portfolio exposures. If images are provided, combine the visual signals with the Excel data. Provide concise institutional insights, risk drivers, trading implications, hedge ideas, confirmation signals, and invalidation triggers.",
)
@应用.entrypoint
def multimodal_data_processor(payload, context):
提示词 = payload.get("提示词", "Analyze the provided data.")
excel = base64.b64decode(payload.get("excel_data", ""))
image = base64.b64decode(payload.get("image_data", ""))
response = agent([
{
"document": {
"format": "xlsx",
"name": "excel_data",
"source": {"bytes": excel},
}
},
{
"image": {
"format": "png",
"source": {"bytes": image},
}
},
{
"text": f"{提示词}\nAnalyze both data sources and provide combined insights."
},
])
return response.message["content"][0]["text"]
if __name__ == "__main__":
应用.run()
运行AgentCore Runtime大负载处理
import base64, uuid
from IPython.display import Markdown, display
def b64(path):
with open(path, "rb") as f:
return base64.b64encode(f.read()).decode()
payload = {
"提示词": "Analyze the FSI Excel data and chart. Focus on bond yields, credit spreads, sovereign risk, FX pressure, liquidity, capital flows, portfolio exposures, risk drivers, trading implications, hedge ideas, confirmation signals, and invalidation triggers.",
"excel_data": b64("large_credit_spreads_data.xlsx"),
"image_data": b64("credit_spreads_chart.png"),
}
res = agentcore_runtime.invoke(payload, session_id=str(uuid.uuid4()))
display(Markdown("".join(res["response"])))
演示：带网关的 AgentCore Runtime
AgentCore Runtime输入：
您是资深机构市场专家，专门研究信用利差扩大、主权债务风险重新定价和跨资产风险轮换。您可以执行数学计算。提供有关主权风险、中美国债收益率利差、信用利差、久期、FX 对冲、流动性、资本流向、投资组合风险敞口和防御性定位的简明交易观点。当提供 Excel 文件或图像时，结合结构化数据和视觉信号来提供见解、交易影响、对冲想法、确认信号和失效触发器。
具有网关输出 1 的 AgentCore Runtime：
美元流动性周期 -> 中国工业需求引擎 -> 欧洲主权信用破裂 -> 全球风险溢价迁移 -> 有纪律的多资产轮动
（1）社会压力、失业、紧缩、都市化、不平等、移民和信任崩溃不是背景噪音，而是未来风险溢酬。
罗伯特席勒的非理性繁荣的核心教训：
(1)过高的估值可以比理性持续更久，但不可能永远持续。
(2)当价格与收益脱节时，信念本身就变成了一种资产，而信念可能会突然消失。
(3) 真正的优势在于在达成共识之前完成有纪律的迁移。
Raghuram Rajan，《金融发展是否使世界风险更大？ 》 Markus Brunnermeier，《解读 2007-2008 年流动性与信贷紧缩》
（1）美国通过低利率、房地产泡沫、次级抵押贷款、证券化、杠杆和影子银行重建风险偏好。
托马斯·弗里德曼的《世界是平的》
（一）中国成为全球需求引擎。
（二）钢铁、煤炭、铜、铝、港口、出口、都市化、国营企业上市、基础建设。
(3) 全球化的世界，但更沉重，承载着煤烟、资源和资产负债表。
席勒、阿克洛夫和席勒的《动物精神》塔勒布的《黑天鹅》
(1) 市场心理、叙事和极端冲击不是次要问题，而是定价力量。
莱因哈特和罗格夫的这次不同
戈顿被「看不见的手」打了一巴掌
(1)银行危机是资产负债表危机，而不是普通的衰退。
（2）降低杠杆，买入流动性：美元现金、美国国债、黄金、高等级信用债。
布兰查德、戴尔阿里恰和毛罗对宏观经济政策的反思
(1)真正的赢家是轮替纪律。
具有网关输出 2 的 AgentCore Runtime：
三层轮动结构：生存资本、景气循环资本、冲击资本
生存资本：
（一）配置美元现金、短期美国国债、高等级信用债、黄金。
景气循环资本：
(1)在主导宏观引擎之间轮换。如果美国企业获利、就业、房屋和信贷利差改善，则购买具有实际现金流的美国股票、资本重组金融股、投资等级信贷和科技股。
(2) 如果中国的PMI、铜价、煤炭价格、铁矿石、亚洲出口和基础设施订单上涨，则买入与中国相关的周期股、资源股、航运和高贝塔亚洲股票。
（3）如果欧洲因PIIGS利差扩大、银行压力上升、欧元贬值而走弱，则减少欧元信贷风险敞口，避开弱势银行，持有德国国债或美元资产。
冲击资本：
(1) 买入保护：黄金、波动性工具、信用违约掉期对冲、美元多头头寸及现金储备。
具有网关输出 3 的 AgentCore Runtime：
信用关键字：
(1) 成长、流动性、偿债能力、货币实力和政治动员能力。
信用风险关键字：
（一）资金流动性、信用利差迁移、货币错配、政策反应函数、头寸拥挤、相关性冲击。
(2)美国拥有Federal储备、储备货币地位、深度资本市场和量化宽松。
（3）中国有WTO红利、国营银行、基础建设需求、资本管制和政策动员能力。它也存在大宗商品依赖、地方政府债务、产能过剩、环境压力等问题。
(4)欧洲货币坚挺。
大到不能倒、大做空、万恶之源和危机经济学
(1) 可见的价格不如隐藏的杠杆那么危险。
具有网关最终输出的 AgentCore Runtime：
战术套利和凸性交易
（1）宏观流动性过剩→房地产信贷扩张→结构性信贷错误定价→杠杆凸性套利→纪律可控退出。
(2) 他不仅看到了「次级房贷」。他看到家庭债务转化为现金流，现金流转化为证券，证券转化为评级，评级转化为资产负债表杠杆。
(3) 这里的心理关键字是控制贪婪：不是盲目乐观，而是认知到资本被迫沿着风险曲线向外流动。
（4）房价上涨，违约相关性较低，信用利差收窄，AAA评级结构性产品收益率高于同评级公司债。
(5) 当政策、人口、金融工程和投资者约束都指向同一个方向时，交易可能会很拥挤，但在它变得致命之前，往往是有利可图的。
（六）持有抵押贷款信贷机器中的优质部分，以低成本融资，对冲明显的尾部风险。
(7) 交易者将资金配置到高级和夹层抵押贷款支持证券、选定的 CDO 档次以及流动性更强的信用违约掉期指数。
(8) 避免毒性最强的无文件贷款池，并优先选择具有超额利差、超额抵押、触发机制和较短加权平均期限的结构。
(9) 了解银行、保险公司、退休基金、对冲基金和外汇储备管理者都在购买同样的收益率幻象，从而在信贷周期破裂之前创造了可交易的价格趋势。
杠铃学分分配
(1) 流动性较强的高阶结构性信用额度、相对价值信用违约掉期头寸及现金储备，以满足保证金压力。
(2) 增加 AAA 和 AA 抵押贷款部分的风险敞口，因为这些资产相对于企业信贷仍提供更高的利差补偿。
(3) 通过信用违约掉期购买保护，以对冲较弱的抵押贷款指数、高风险贷款发起人和高杠杆金融机构。
交易关键字：
(1)房价升值、拖欠移转、ABX指数利差、回购融资折扣。
(2) 如果房价上涨缓慢，拖欠率上升，融资折扣扩大，交易必须从获取套利收入转向保护资本。
(3) 利用周期后期结构性信贷购买压力，持有融资脆弱性低于同业的头寸，保留流动性对冲，并在被迫卖方主导市场之前退出。
（4）买入高级套利资产，对冲系统性冲击，准备反向退出。交易不得用作单向杠杆。
（5）结构性从属保护较强的地方收取保费，承保质量明显恶化的地方购买保护。
风险关键字：
(1)相关性冲击、资金螺旋、模型误差、交易对手错向风险、流动性错觉。
(2) 证券化可以将局部抵押贷款问题转化为全球资产负债表问题。
（3）次级房贷出现多元化，但如果全国承保标准同时下降，多元化就会失败。
(4) 信用违约掉期看似转移了风险，但也可能集中交易对手风险并造成虚假的安全感。
（5）利用市场人群对收益的渴望，但绝不依赖人群的怜悯。
分级评级关键字：
(1)房价、违约相关性、回收率、再融资能力、市场流动性。
(2)持有透明抵押品。
（3）持有较强次级保护和退出流动性的短期工具。
第 11 部分.Amazon Bedrock AgentCore 网关
引入 AgentCore Gateway 作为托管 MCP 整合层，将 API、Lambda、Smithy 和 OpenAPI 定义转换为安全、代理可访问的企业工具，具有身份验证内置功能、可观测性和治理控制功能。
网关职责：
- 将 API、Lambda 功能和现有服务转换为 MCP 相容工具。
- 支持 OpenAPI、Smithy 和 AWS Lambda。
- 在完全托管的服务中提供入口身份验证和出口身份验证。
- 提供与 Salesforce、Slack、Jira、Asana 和 Zendesk 的一键整合。
组件摘要：
- Lambda 功能：业务逻辑
- 授权者：OAuth/OIDC配置
- Smithy：强型工具
- 翻译引擎：将 MCP JSON-RPC 转换为 REST 或 Lambda 负载
- 语义搜索索引：带有嵌入的目录工具
- 可观测性管道：跟踪 CloudWatch 中的每次呼叫延迟、状态、错误分类和目标级呼叫计数
构建网关和 MCP 端点 URL：
呼叫CreateMcpGateway、OAuth授权者配置和IAM角色。
目标注册：
附加 Lambda、OpenAPI 或 Smithy 定义。
代理呼叫：
- 代理从 OIDC 提供者获取访问令牌。
- 智能体使用工具/列表或工具/呼叫来呼叫 MCP 端点。
- 使用执行角色使用 SigV4 对请求进行签署。
- 翻译引擎重写负载以符合目标协议。
语义工具选择：
- 智能体向内置搜索工具发送工具/呼叫。
- 网关查询其嵌入索引并返回符合工具的排名列表。
第 12 部分. 将 Lambda 功能转换为 MCP 工具
概述如何构建 Lambda 支持的 MCP 工具、IAM 角色、Cognito 授权、网关目标、工具架构和凭证提供程序，以便通过代理在安全的 AWS 生产环境和工作流中实现一致、受控的业务逻辑。
将 lambda 转换为 MCP 工具
步骤1.使用uv_install()安装所需的AWS SDK软件包。
步骤 2. 从 AWS_DEFAULT_REGION 配置 REGION。
步骤 3. 配置 LAMBDA_ZIP_FILE、GATEWAY_NAME 和 TARGET_NAME 等常数。
步骤 4. 使用 create_lambda_function() 构建示例 Lambda 函数。
步骤 5. 使用 create_gateway_iam_role() 构建 AgentCore Gateway IAM 角色。
步骤6.使用setup_cognito_authorization()构建Cognito授权资源。
步骤7.使用create_agentcore_gateway()构建MCP AgentCore Gateway。
步骤 8. 使用 Cognito 自定义 JWT 授权者配置 auth_config。
步骤9.使用create_lambda_gateway_target()将Lambda注册为MCP工具。
步骤 10. 定义 get_order_tool 和 update_order_tool 的工具架构。
步骤 11. 将 credential_config 与 GATEWAY_IAM_ROLE 结合使用。
步骤 12. 使用 main() 运行端对端网关配置。
步骤 13. 返回 Lambda、Cognito、网关和目标构建输出。
import utils
import os
import sys
import shutil
import subprocess
# Step 1: Ensure uv, boto3, and botocore are installed
def ensure_uv_installed():
if shutil.which("uv") is None:
subprocess.check_call([sys.executable, "-m", "pip", "install", "uv"])
def uv_install(*packages):
ensure_uv_installed()
uv_path = shutil.which("uv")
if uv_path is None:
raise RuntimeError("uv installation failed or uv is not available in PATH.")
subprocess.check_call([uv_path, "pip", "install", "--upgrade", *packages])
uv_install("botocore", "boto3")
# Step 2: Import AWS SDK dependencies
import boto3
from botocore.exceptions import ClientError
import utils
# Step 3: Configure constants
REGION = os.environ.get("AWS_DEFAULT_REGION")
if not REGION:
raise EnvironmentError("AWS_DEFAULT_REGION environment variable is not set.")
LAMBDA_ZIP_FILE = "lambda_function_code.zip"
GATEWAY_ROLE_NAME = "sample-lambdagateway"
USER_POOL_NAME = "sample-agentcore-gateway-pool"
RESOURCE_SERVER_ID = "sample-agentcore-gateway-id"
RESOURCE_SERVER_NAME = "sample-agentcore-gateway-name"
CLIENT_NAME = "sample-agentcore-gateway-client"
SCOPES = [
{
"ScopeName": "gateway:read",
"ScopeDescription": "Read access"
},
{
"ScopeName": "gateway:write",
"ScopeDescription": "Write access"
}
]
GATEWAY_NAME = "TestGWforLambda"
GATEWAY_DESCRIPTION = "AgentCore Gateway with AWS Lambda target type"
TARGET_NAME = "LambdaUsingSDK"
TARGET_DESCRIPTION = "Lambda Target using SDK"
# Step 4: Create sample AWS Lambda function
def create_lambda_function():
lambda_resp = utils.create_gateway_lambda(LAMBDA_ZIP_FILE)
if lambda_resp is None:
raise RuntimeError("Lambda function creation returned None.")
if lambda_resp.get("exit_code") == 0:
return lambda_resp.get("lambda_function_arn")
raise RuntimeError(
lambda_resp.get(
"lambda_function_arn",
"Lambda function creation failed."
)
)
# Step 5: Create IAM role for AgentCore Gateway
def create_gateway_iam_role():
role_response = utils.create_agentcore_gateway_role(GATEWAY_ROLE_NAME)
return role_response["Role"]["Arn"]
# Step 6: Create Amazon Cognito resources for inbound authorization
def setup_cognito_authorization():
cognito = boto3.client("cognito-idp", region_name=REGION)
user_pool_id = utils.get_or_create_user_pool(
cognito,
USER_POOL_NAME
)
utils.get_or_create_resource_server(
cognito,
user_pool_id,
RESOURCE_SERVER_ID,
RESOURCE_SERVER_NAME,
SCOPES
)
client_id, client_secret = utils.get_or_create_m2m_client(
cognito,
user_pool_id,
CLIENT_NAME,
RESOURCE_SERVER_ID
)
cognito_discovery_url = (
f"https://cognito-idp.{REGION}.amazonaws.com/"
f"{user_pool_id}/.well-known/openid-configuration"
)
return client_id, client_secret, cognito_discovery_url
# Step 7: Create AgentCore Gateway with Amazon Cognito authorizer
def create_agentcore_gateway(role_arn, client_id, cognito_discovery_url):
gateway_client = boto3.client(
"bedrock-agentcore-control",
region_name=REGION
)
auth_config = {
"customJWTAuthorizer": {
"allowedClients": [client_id],
"discoveryUrl": cognito_discovery_url
}
}
create_response = gateway_client.create_gateway(
name=GATEWAY_NAME,
roleArn=role_arn,
protocolType="MCP",
authorizerType="CUSTOM_JWT",
authorizerConfiguration=auth_config,
description=GATEWAY_DESCRIPTION
)
gateway_id = create_response["gatewayId"]
gateway_url = create_response["gatewayUrl"]
return gateway_client, gateway_id, gateway_url, create_response
# Step 8: Register Lambda target as MCP tools
def create_lambda_gateway_target(gateway_client, gateway_id, lambda_arn):
lambda_target_config = {
"mcp": {
"lambda": {
"lambdaArn": lambda_arn,
"toolSchema": {
"inlinePayload": [
{
"name": "get_order_tool",
"description": "tool to get the order",
"inputSchema": {
"type": "object",
"properties": {
"orderId": {
"type": "string"
}
},
"required": ["orderId"]
}
},
{
"name": "update_order_tool",
"description": "tool to update the orderId",
"inputSchema": {
"type": "object",
"properties": {
"orderId": {
"type": "string"
}
},
"required": ["orderId"]
}
}
]
}
}
}
}
credential_config = [
{
"credentialProviderType": "GATEWAY_IAM_ROLE"
}
]
response = gateway_client.create_gateway_target(
gatewayIdentifier=gateway_id,
name=TARGET_NAME,
description=TARGET_DESCRIPTION,
targetConfiguration=lambda_target_config,
credentialProviderConfigurations=credential_config
)
return response
# Step 9: Run end-to-end setup
def main():
lambda_arn = create_lambda_function()
gateway_role_arn = create_gateway_iam_role()
client_id, client_secret, cognito_discovery_url = setup_cognito_authorization()
gateway_client, gateway_id, gateway_url, gateway_response = create_agentcore_gateway(
role_arn=gateway_role_arn,
client_id=client_id,
cognito_discovery_url=cognito_discovery_url
)
target_response = create_lambda_gateway_target(
gateway_client=gateway_client,
gateway_id=gateway_id,
lambda_arn=lambda_arn
)
return {
"lambda_arn": lambda_arn,
"gateway_role_arn": gateway_role_arn,
"client_id": client_id,
"client_secret": client_secret,
"cognito_discovery_url": cognito_discovery_url,
"gateway_id": gateway_id,
"gateway_url": gateway_url,
"gateway_response": gateway_response,
"target_response": target_response
}
if __name__ == "__main__":
main()
第 13 部分. 从 Strands Agent 呼叫 AgentCore Gateway
演示如何将 Strands Agent 与经过 SigV4 验证的 MCP 传输、临时 STS 凭证、工具探索、直接工具呼叫以及通过 AgentCore Gateway 进行代理驱动的执行结合使用，以实现安全的企业整合和受控自动化场景。
通过 Strands Agent 致电 Bedrock AgentCore Gateway
步骤 1. 配置 AgentCore Gateway SigV4 验证的服务。
步骤 2. 使用 BedrockModel 初始化模型。
步骤 3. 使用 create_streamable_http_transport_sigv4() 构建经过 IAM 验证的 MCP 传输。
步骤 4. 使用 get_full_tools_list() 从网关检索所有 MCP 工具。
步骤5.使用call_tool_sync()直接呼叫MCP工具。
步骤6.使用run_agent()构建MCPClient并将工具载入到Strands Agent中。
步骤7.使用MCP工具使用agent()询问订单状态。
步骤8.使用assume_gateway_invoke_role()获取临时STS凭证。
步骤9.从环境变量读取gateway_url、role_arn和region。
步骤 10. 使用网关 URL 和假设的 IAM 凭证呼叫 run_agent()。
pip3 install --upgrade strands-agents strands-agents-tools boto3 botocore
import os
import time
import boto3
from botocore.credentials import Credentials
from strands import Agent
from strands.models import BedrockModel
from strands.tools.mcp.mcp_client import MCPClient
from mcp.client.streamable_http import streamablehttp_client
from streamable_http_sigv4 import streamablehttp_client_with_sigv4
SERVICE = "bedrock-agentcore"
model = BedrockModel(
model_id="us.amazon.nova-pro-v1:0",
temperature=0.7,
)
def create_streamable_http_transport_sigv4(
mcp_url, key, secret, session_token, service_name, aws_region
):
return streamablehttp_client_with_sigv4(
url=mcp_url,
credentials=Credentials(
access_key=key,
secret_key=secret,
token=session_token,
),
service=service_name,
region=aws_region,
)
def get_full_tools_list(client):
tools, pagination_token = [], None
while True:
result = client.list_tools_sync(pagination_token=pagination_token)
tools.extend(result)
pagination_token = result.pagination_token
if pagination_token is None:
return tools
def call_tool_sync(client, tool_id, tool_name, parameters=None):
response = client.call_tool_sync(
tool_use_id=tool_id,
name=tool_name,
arguments=parameters or {},
)
if hasattr(response, "results") and response.results:
return response.results
if hasattr(response, "output") and response.output:
return response.output
if hasattr(response, "content"):
return response.content
return response
def run_agent(mcp_url, key, secret, session_token, service_name, aws_region):
mcp_client = MCPClient(
lambda: create_streamable_http_transport_sigv4(
mcp_url,
key,
secret,
session_token,
service_name,
aws_region,
)
)
with mcp_client:
tools = get_full_tools_list(mcp_client)
print(f"Found the following tools: {[tool.tool_name for tool in tools]}")
print(f"First tool name: {tools[0].tool_name}")
agent = Agent(model=model, tools=tools)
print(f"Tools loaded in the agent are: {agent.tool_names}")
agent_response = agent(
"Check the order status for order id 123 and show me the exact response from the tool"
)
print("Agent response:")
print(agent_response)
result = call_tool_sync(
mcp_client,
"get-order-id-123-call-1",
tools[0].tool_name,
{"orderId": "123"},
)
print("Tool Call result:")
try:
print(result["content"][0]["text"])
except Exception:
print(result)
def assume_gateway_invoke_role(role_arn):
return boto3.client("sts").assume_role(
RoleArn=role_arn,
RoleSessionName="invoke_mcp_session",
DurationSeconds=3600,
)["Credentials"]
def main():
gateway_url = os.environ["AGENTCORE_GATEWAY_URL"]
role_arn = os.environ["AGENTCORE_GATEWAY_INVOKE_ROLE_ARN"]
region = os.environ.get("AWS_DEFAULT_REGION") or os.environ["AWS_REGION"]
creds = assume_gateway_invoke_role(role_arn)
run_agent(
gateway_url,
creds["AccessKeyId"],
creds["SecretAccessKey"],
creds["SessionToken"],
SERVICE,
region,
)
if __name__ == "__main__":
main()
export AWS_DEFAULT_REGION=us-east-1
export AGENTCORE_GATEWAY_URL="https://your-gateway-url"
export AGENTCORE_GATEWAY_INVOKE_ROLE_ARN="arn:aws:iam::123456789012:role/your-gateway-invoke-role"
python agentcore_gateway_strands_mcp.py
第 14 部分.AgentCore Gateway 语义搜索
解释 Gateway 如何通过嵌入对工具元数据进行索引并公开语义搜索，以便智能体发现相关工具，而无需将每个模式加载到提示中，从而显著降低延迟成本和复杂性。
AgentCore Gateway 语义搜索核心元件
AgentCore Gateway：
聪明地发现并仅选择与多个 Lambda 支持的目标最相关的工具
Strands Agent：
与框架无关的代理层，使用 Amazon Bedrock 模型进行自然语言理解、推理和编排任务。
AgentCore Gateway：
集中式工具管理中心使代理能够有效率、安全地发现、选择和呼叫企业工具。
Lambda 目标：
依类分组的 AWS Lambda 功能集合，代表不同的企业功能和整合工作流。
语义搜索引擎：
智慧发现层将代理意图与相关工具相匹配，而无需处理详尽的列表或列表。
警卫：
管理 OAuth 授权，确保只有经过验证的用户和智能体才能访问批准的工具和资源。
译者：
将 MCP 和其他代理协议请求转换为 API 呼叫和 Lambda 呼叫，无需服务器。
作曲家：
将多个 API、功能和工具组合到一个 MCP 端点中，代理可以无缝、有效率地使用。
钥匙圈：
为每个工具注入正确的凭证，使代理人能够安全可靠地使用受保护的服务。
研究员：
根据场景搜索工具库存，选择相关工具，显著减少提示大小、延迟和运营成本。
基础设施经理：
为无服务器网关基础架构提供可观测性和审计功能，减轻开发者和团队的运营负担。
向量存储语义搜索流程
创建向量商店：
Gateway 自动为语义工具搜索功能提供无服务器、完全托管的向量存储。
向量商店已填充：
Gateway 使用嵌入来索引来自 Lambda 目标、JSON 定义和 OpenAPI 架构的工具元数据。
提供的搜索工具：
Gateway 新增了 x-amz-bedrock-agentcore-search，这是一个 MCP 工具，用于跨可用 Gateway 工具进行语义发现。
语义查询参数：
搜索工具接受单一查询参数并返回按场景相关性排名的工具。
元数据匹配：
搜索将查询与工具名称、描述、输入模式和输出模式进行比较，以实现准确发现。
构建 AgentCore Gateway 语义搜索
- Amazon Cognito 作为身份提供者 (IdP)
- AWS Lambda 作为目标
- AWS IAM 用于出站身份验证
步骤 1：配置网关名称
GATEWAY_NAME 存储 AgentCore 网关名称。
第 2 步：定义工具架构
TOOL_SCHEMA 定义 sovereign_debt_risk_repricing。
步骤 3：构建AWS会话
session = boto3.Session() 构建 AWS 会话。
步骤 4：构建AgentCore客户端
agentcore_client 连接到 bedrock-agentcore-control。
步骤 5：构建网关
create_gateway(cognito, gateway_role_arn) 构建 MCP 网关。
步骤 6：配置JWT认证
cognito["client_id"] 和 cognito["discovery_url"] 配置身份验证。
步骤 7：配置MCP搜索
protocolConfiguration 启用语义 MCP 搜索。
步骤 8：返回网关ID
create_gateway() 返回 response["gatewayId"]。
第 9 步：构建网关目标
create_gateway_target(gateway_id, lambda_arn) 构建 Lambda 目标。
步骤 10：附加 Lambda ARN
lambda_arn 将 Lambda 连接到网关目标。
第 11 步：附加工具架构
TOOL_SCHEMA 作为 inlinePayload 传递。
步骤 12：返回目标ID
create_gateway_target() 返回 response["targetId"]。
第 13 步：构建Lambda函数
utils.create_gateway_lambda() 返回 lambda_resp。
第 14 步：配置Cognito
utils.setup_cognito_user_pool() 返回 cognito。
第 15 步：获取不记名令牌
utils.get_bearer_token() 返回 bearer_token。
第 16 步：创建IAM角色
utils.create_gateway_iam_role() 返回 gateway_role_arn。
步骤 17：构建网关资源
gateway_id = create_gateway(...)。
第 18 步：等待传播
time.sleep(5) 等待 AWS 准备就绪。
第 19 步：构建目标资源
target_id = create_gateway_target(...)。
第 20 步：列出目标
targets_response 存储网关目标列表。
步骤 21：返回输出
main() 返回 ID、目标和令牌。
import time
import boto3
import utils
GATEWAY_NAME = "gateway-sdr-repricing"
TOOL_SCHEMA = [
{
"name": "sovereign_debt_risk_repricing",
"description": "Sovereign debt risk repricing",
"inputSchema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": (
"Question, analysis request, or market context "
"for sovereign debt risk repricing."
),
}
},
"required": ["query"],
},
}
]
session = boto3.Session()
agentcore_client = session.client("bedrock-agentcore-control")
def create_gateway(cognito, gateway_role_arn):
response = agentcore_client.create_gateway(
name=GATEWAY_NAME,
roleArn=gateway_role_arn,
authorizerType="CUSTOM_JWT",
description="AgentCore Gateway Sovereign Debt Risk Repricing",
protocolType="MCP",
authorizerConfiguration={
"customJWTAuthorizer": {
"allowedClients": [cognito["client_id"]],
"discoveryUrl": cognito["discovery_url"],
}
},
protocolConfiguration={
"mcp": {
"searchType": "SEMANTIC",
"supportedVersions": ["2025-03-26"],
}
},
)
return response["gatewayId"]
def create_gateway_target(gateway_id, lambda_arn):
response = agentcore_client.create_gateway_target(
gatewayIdentifier=gateway_id,
name="SDRRepricingTools",
description="Sovereign Debt Risk Repricing Tools",
targetConfiguration={
"mcp": {
"lambda": {
"lambdaArn": lambda_arn,
"toolSchema": {
"inlinePayload": TOOL_SCHEMA,
},
}
}
},
credentialProviderConfigurations=[
{
"credentialProviderType": "GATEWAY_IAM_ROLE",
}
],
)
return response["targetId"]
def main():
lambda_resp = utils.create_gateway_lambda(
"calc/lambda_function_code.zip",
lambda_function_name="sdr_repricing_lambda_gateway",
)
cognito = utils.setup_cognito_user_pool()
bearer_token = utils.get_bearer_token(
client_id=cognito["client_id"],
username="testuser",
password="MyPassword123!",
)
gateway_role_arn = utils.create_gateway_iam_role(
lambda_arns=[lambda_resp["lambda_function_arn"]]
)
gateway_id = create_gateway(
cognito=cognito,
gateway_role_arn=gateway_role_arn,
)
target_id = create_gateway_target(
gateway_id=gateway_id,
lambda_arn=lambda_resp["lambda_function_arn"],
)
targets_response = agentcore_client.list_gateway_targets(
gatewayIdentifier=gateway_id
)
return {
"gateway_id": gateway_id,
"target_id": target_id,
"targets_response": targets_response,
"bearer_token": bearer_token,
}
if __name__ == "__main__":
main()
AgentCore Gateway 语义搜索的搜索工具
步骤 1：get_gateway_endpoint使用agentcore_client和gateway_id检索gateway_endpoint，返回MCP客户端连接的gatewayUrl。
步骤 2：main通过utils.get_bearer_token获取jwt_token，使用cognito_response、用户名称和密码进行网关验证。
步骤 3：MCPClient 使用 gateway_endpoint 和 jwt_token 标头开启客户端，从而启用 JSON-RPC agent_ready_tools 构建。
步骤 4：get_all_agent_tools_from_mcp_endpoint对next_cursor进行分页，将每个工具转换为MCPTool和MCPAgentTool。
步骤 5：第二个MCPClient再次开启客户端，通过Strands list_tools_sync分页隔离mcp_tools检索。
步骤 6：get_all_mcp_tools_from_mcp_client 循环 pagination_token，用每个返回的页面扩充工具直到完成。
步骤 7：tool_search用search_query构建tool_params，通过invoke_gateway_tool呼叫x_amz_bedrock_agentcore_search。
步骤 8：invoke_gateway_tool 使用 jwt_token 将 tool_params 发布到 gateway_endpoint，从工具/呼叫返回 JSON-RPC 响应。
步骤 9：main 返回 gateway_endpoint、jwt_token、agent_ready_tools、mcp_tools、semantic_search_results 和经过的搜索秒数。
import time
import requests
from strands.tools.mcp import MCPClient
from strands.tools.mcp.mcp_agent_tool import MCPAgentTool
from mcp import Tool as MCPTool
from mcp.client.streamable_http import streamablehttp_client
def get_gateway_endpoint(agentcore_client, gateway_id):
return agentcore_client.get_gateway(
gatewayIdentifier=gateway_id
)["gatewayUrl"]
def invoke_gateway_tool(gateway_endpoint, jwt_token, tool_params):
return requests.post(
gateway_endpoint,
json={
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": tool_params,
},
headers={
"Authorization": f"Bearer {jwt_token}",
"Content-Type": "应用lication/json",
},
).json()
def get_all_agent_tools_from_mcp_endpoint(gateway_endpoint, jwt_token, client):
tools_list = []
next_cursor = None
while True:
result = requests.post(
gateway_endpoint,
json={
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {"cursor": next_cursor} if next_cursor else {},
},
headers={
"Authorization": f"Bearer {jwt_token}",
"Content-Type": "应用lication/json",
},
).json()["result"]
for tool in result["tools"]:
tools_list.应用end(
MCPAgentTool(
MCPTool(
name=tool["name"],
description=tool.get("description", ""),
inputSchema=tool.get("inputSchema", {}),
),
client,
)
)
next_cursor = result.get("nextCursor")
if next_cursor is None:
return tools_list
def get_all_mcp_tools_from_mcp_client(client):
tools = []
pagination_token = None
while True:
page = client.list_tools_sync(
pagination_token=pagination_token
)
tools.extend(page)
pagination_token = page.pagination_token
if pagination_token is None:
return tools
def tool_search(gateway_endpoint, jwt_token, query):
return invoke_gateway_tool(
gateway_endpoint=gateway_endpoint,
jwt_token=jwt_token,
tool_params={
"name": "x_amz_bedrock_agentcore_search",
"arguments": {
"query": query
},
},
)["result"]["structuredContent"]["tools"]
def main(
agentcore_client,
utils,
cognito_response,
gateway_id,
username="user",
password="pwd",
search_query="search credit spread widening and extra 2 related tools",
):
gateway_endpoint = get_gateway_endpoint(
agentcore_client=agentcore_client,
gateway_id=gateway_id,
)
jwt_token = utils.get_bearer_token(
client_id=cognito_response["client_id"],
username=username,
password=password,
)
with MCPClient(
lambda: streamablehttp_client(
gateway_endpoint,
headers={
"Authorization": f"Bearer {jwt_token}"
},
)
) as client:
agent_ready_tools = get_all_agent_tools_from_mcp_endpoint(
gateway_endpoint=gateway_endpoint,
jwt_token=jwt_token,
client=client,
)
with MCPClient(
lambda: streamablehttp_client(
gateway_endpoint,
headers={
"Authorization": f"Bearer {jwt_token}"
},
)
) as client:
mcp_tools = get_all_mcp_tools_from_mcp_client(client)
start_time = time.time()
semantic_search_results = tool_search(
gateway_endpoint=gateway_endpoint,
jwt_token=jwt_token,
query=search_query,
)
return {
"gateway_endpoint": gateway_endpoint,
"jwt_token": jwt_token,
"agent_ready_tools": agent_ready_tools,
"mcp_tools": mcp_tools,
"semantic_search_results": semantic_search_results,
"semantic_search_elapsed_seconds": time.time() - start_time,
}
第 15 部分：主权风险交易方法
围绕着成长、流动性、偿付能力、货币实力、政策反应函数、资本流动、信用利差迁移、FX 压力以及机构主权风险决策和对冲的跨资产轮换构建市场方法论。
主权风险交易方法论以成长、流动性、偿付能力、货币实力、政治动员能力、融资流动性、信用利差迁徙、货币错配、政策反应函数、拥挤交易和相关性冲击为中心。
实际的交易结构使用三层：
- 生存资本
- 景气循环资本
- 冲击资本
生存资本可以防止强制清算。景气循环资本跟随主导的宏观引擎。当信贷、流动性或政策制度崩溃时，冲击资本会购买保护。
第 16 部分：历史危机背景
使用长期萧条、霸菱危机、经济大恐慌、石油冲击、亚洲金融风暴和 LTCM 等历史危机来解释流动性重新定价、杠杆传染和防御性投资组合轮换纪律教训。
历史危机背景包括：
- 长期萧条
- 霸菱危机
- 经济大恐慌
- 石油冲击
- 亚洲金融风暴
- LTCM
这些事件解释了流动性重新定价、杠杆传染、资本流动逆转、防御性投资组合轮换以及在达成共识之前迁移资本的必要性。