使用 Amazon Bedrock AgentCore 建立主權風險交易代理，分析殖利率差、FX 避險與債務重新定價
Amazon Bedrock AgentCore 協助團隊建置生產等級的主權風險交易代理，分析政府公債殖利率差、信用價差擴大、債務重新定價、FX 波動、流動性壓力與防禦性資產輪動。
金融服務系列
- 第 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_reaction_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 BedrockAgentCoreApp
from strands.models import BedrockModel
app = BedrockAgentCoreApp()
model = BedrockModel(
model_id="amazon.nova-pro-v1:0"
)
agent = Agent(
model=model,
tools=[calculator],
system_prompt="You're a credit spread widening and sovereign debt risk repricing expert. You can do math calculations."
)
@app.entrypoint
def strands_agent_bedrock(payload):
user_input = payload.get("prompt")
response = agent(user_input)
return response.message["content"][0]["text"]
if __name__ == "__main__":
app.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({"prompt": prompt}).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_reaction_function() -> str:
"""Analyze policy reaction function"""
return "Policy reaction 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": "application/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 BedrockAgentCoreApp
from strands.models import BedrockModel
app = BedrockAgentCoreApp()
@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_prompt="You're a credit spread widening and sovereign debt risk repricing expert. You can do math calculations." ,
)
@app.entrypoint
async def strands_agent_bedrock_streaming(payload):
try:
async for event in agent.stream_async(payload.get("prompt")):
if "data" in event:
yield event["data"]
except Exception as e:
yield {"error": str(e), "type": "stream_error"}
if __name__ == "__main__":
app.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({
"prompt": "Focus specifically on the US-China government bond yield spread and what it implies for the next position action. 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 action 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 appetite. Finally, give clear invalidation points that would require changing the position quickly."
})
invoke_response
使用 AgentCore Runtime 串流回應
處理 Server-Sent Events 格式。
使用boto3進行串流
import boto3, json
prompt = "Analyze the US-China government bond yield spread for next-position action. 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 appetite."
res = boto3.client("bedrock-agentcore", region_name=region).invoke_agent_runtime(
agentRuntimeArn=launch_result.agent_arn,
qualifier="DEFAULT",
payload=json.dumps({"prompt": prompt}),
)
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
prompt = "Analyze the US-China government bond yield spread for next-position action. 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 appetite."
res = boto3.client("bedrock-agentcore", region_name=region).invoke_agent_runtime(
agentRuntimeArn=launch_result.agent_arn,
qualifier="DEFAULT",
payload=json.dumps({"prompt": prompt}),
)
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 BedrockAgentCoreApp
from strands.models import BedrockModel
app = BedrockAgentCoreApp()
@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_prompt= “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",
}
@app.entrypoint
def strands_agent_bedrock_handling_context(payload, context):
name = users[payload["user_id"]]
prompt = f"My name is {name}. Request: {payload.get('prompt')}\nSession: {context.session_id}. Analyze the US-China government bond yield spread for next-position action. 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 action, hedge options, confirmation signals, and invalidation triggers. Please acknowledge my name and assist."
return agent(prompt).message["content"][0]["text"]
if __name__ == "__main__":
app.run()
建立AgentCore Runtime工作階段管理的工作階段
import uuid
from IPython.display import Markdown, display
session_id = str(uuid.uuid4())
invoke_response = agentcore_runtime.invoke(
{
"prompt": "Analyze the US-China government bond yield spread for next-position action. 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 appetite.",
"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 BedrockAgentCoreApp
app = BedrockAgentCoreApp()
agent = Agent(
model=BedrockModel(
model_id="amazon.nova-pro-v1:0",
max_tokens=16000,
),
system_prompt="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.",
)
@app.entrypoint
def multimodal_data_processor(payload, context):
prompt = payload.get("prompt", "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"{prompt}\nAnalyze both data sources and provide combined insights."
},
])
return response.message["content"][0]["text"]
if __name__ == "__main__":
app.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 = {
"prompt": "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": "application/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": "application/json",
},
).json()["result"]
for tool in result["tools"]:
tools_list.append(
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
這些事件解釋了流動性重新定價、槓桿傳染、資本流動逆轉、防禦性投資組合輪換以及在達成共識之前遷移資本的必要性。