← Financial Cloud Cloud Cloud Club · 建構文章

極點宏觀|Financial Cloud Cloud · 建構文章

使用 AgentCore 與 Strands 建構:Gateway MCP 工具織網開發者工作坊

系列: AgentCore

文章: A1

文章
Kiro 工作坊
01 使用 Kiro 建置:他加祿語學習 App 的提示優先產品設計工作坊
Kiro 工作坊
02 使用 Kiro 建置:他加祿語 學習 App 的教育優先開發技巧工作坊
Kiro 工作坊
03 使用 Kiro 建置:他加祿語 學習 App 的深入開發流程工作坊
Kiro 工作坊
04 使用 Kiro 建置:將 他加祿語 學習 App 在地化為中文變體工作坊
Kiro 工作坊
05 使用 Kiro 建置:他加祿語 卡片的文法與發音補強管線工作坊
Kiro 工作坊
06 使用 Kiro 建置:他加祿語 學習 App 中可審查的獨特額外例句工作坊
Kiro 工作坊
07 與 Kiro 同行:晶圓廠工程健康度 Hook 工作坊
Kiro 工作坊
08 與 Kiro 同行:蝕刻製程視窗風險測試自動化工作坊
Kiro 工作坊
09 與 Kiro 同行:黃光微影漂移風險開發工作坊
Kiro 工作坊
10 工程團隊入門 — 日常工廠值班使用 fab spc drift sync portal
Kiro 工作坊
11 工程團隊附錄 — fab spc drift sync portal 的日常工廠值班使用
Kiro 工作坊
12 Kiro:規格驅動工廠軟體的現場工程工作坊
Kiro 工作坊
13 Kiro:實作 Lab — 從零建置具型別的 Factory Risk Portal
Kiro 工作坊
14 Kiro:工程開發人員的提示、程式碼和型別標準手冊
Kiro 工作坊
15 Kiro:為什麼強 React 提示可以防止型別宣告錯誤啟動
Kiro 工作坊
17 與 Kiro 一起建構:建立工廠自動化入口網站 React UI
Kiro 工作坊
18 與 Kiro 一同建構:打造工廠自動化入口網站背後的自動化分析引擎
Kiro 工作坊
19 與 Kiro 一起實作:將 AI 工廠自動化輔助程式新增至工廠自動化入口網站
Kiro 工作坊
21 Kiro:2 小時專業開發人員工作坊指南
Kiro 工作坊
22 Kiro:從零建置 Fab SPC Drift Synchronization Portal
Kiro 工作坊
23 Kiro:提示詞庫與深度程式碼說明附錄
Kiro 工作坊
30 與 Kiro 一同建構:建立工廠自動化入口網站 UI
Kiro 工作坊
31 與 Kiro 一同建構:打造工廠自動化入口網站背後的自動化分析引擎
Kiro 工作坊
32 與 Kiro 一起實作:為工廠自動化入口網站增添 AI 工廠自動化助理
Kiro 工作坊
33 與 Kiro 一起開發:重建 CME Direct 風格的量化損益排行榜 UI
Kiro 工作坊
34 與 Kiro 一起開發:重建損益排行榜背後的量化分析引擎
Kiro 工作坊
35 與 Kiro 一起建構:適用於量化排行榜的 AWS AI 驅動交易台助理
Kiro 工作坊
36 單頁交易平台 SOP
Kiro 工作坊
AgentCore
A1 使用 AgentCore 與 Strands 建構:Gateway MCP 工具織網開發者工作坊
AgentCore
A2 使用 AgentCore 與 Strands 建構:受治理的多 Agent 風險系統開發者工作坊
AgentCore
A3 使用 AgentCore 與 Strands 建構:執行期主權風險代理人開發者工作坊
AgentCore
模擬考場
E1 用 Vibe Coding 打造多語言 AWS 證照模擬試題上線系統
模擬考場
E2 利用 Vibe Coding 開發技巧打造 AWS 證照模擬練習室
模擬考場
E3 打造靜態 AWS 模擬考場背後的練習引擎
模擬考場
Amazon Q
Q1 Amazon Q:ACM 憑證自動更新的CloudShell優先開發人員工作坊
Amazon Q
Tagalog 練習室
T1 為 AWS Manila Community Day 打造 Tagalog 學習 App:提示詞優先的產品設計
Tagalog 練習室
T2 為 AWS Manila Community Day 打造 Tagalog 學習 App,採用教育優先的開發提示
Tagalog 練習室
T3 為 AWS Manila Community Day 打造 Tagalog 學習 App 的開發流程深度解析
Tagalog 練習室
T4 將 Tagalog 學習 App 在 AWS Manila Community Day 情境中在地化為中文版本
Tagalog 練習室
T5 為 AWS Manila Community Day 打造 Tagalog 卡片打造文法與發音補強流程
Tagalog 練習室
T6 為 AWS Manila Community Day 打造 Tagalog 學習 App 的 Extra Examples 更獨特且可審閱
Tagalog 練習室
路線圖
R1 企業級 Data Analytics Roadmap 一百個深度情境題
路線圖
R2 前端開發路線圖:真實企業場景
路線圖
香港 Community Day
C1 伴隨 AWS Community Day 的香港週末:從雲端技術論壇到維港璀璨夜景
香港 Community Day
C2 講者的奢華週末:分享您的 AWS 故事,讓香港成為您的專屬舞台
香港 Community Day
C3 香港七十二小時:AWS Community Day 講者的極致之旅
香港 Community Day
馬尼拉 Community Day
C4 AWS Community Day Manila:一場連結雲端技術、城市文化與真摯友誼的快樂週末
馬尼拉 Community Day
C5 AWS Community Day Manila:雲端建立者在菲律賓感受最幸福的精神
馬尼拉 Community Day
C6 AWS Community Day Manila:在快樂之城建構、打破、重來,並找到歸屬
馬尼拉 Community Day
C7 菲律賓馬尼拉初次造訪建議
馬尼拉 Community Day
菲律賓 × 香港
C8 菲律賓香港資本市場升級
菲律賓 × 香港
回測
B1 使用 Bedrock AgentCore 與 Strands Agents 建立機構級 Amazon 純做多 (Long-Only) 回測代理
純做多 AMZN 代理:AgentCore、Strands 與可稽核的 Backtrader 帳本。
B2 使用 Backtrader、AgentCore 與 Strands Agents 建立具市況感知能力的 Amazon 部位管理
把市況當成部位控制,而不是圖表註解。
B3 使用 Nasdaq、S&P 500、Dow、AgentCore 與 Strands 建立相對於基準的 Amazon 進出場時機系統
相對 Nasdaq、S&P 500 與道瓊來判斷 AMZN 時機。
B4 使用 Bedrock AgentCore、Strands Agents 與 Backtrader 建立受治理的 Amazon 交易歷史工廠(Trade-History Factory)
把回測做成可稽核的交易歷史工廠。
B5 使用 Bedrock AgentCore 與 Strands Agents 建立智慧代理型 (Agentic) Amazon 回測營運模型 [Part 1]
先建立營運模型,再爭論結果。
B6 為 Amazon 擇時與部位管理建立客製化 Cerebro 程式碼說明 [第 2 部分]
先講 Cerebro 引擎,再講圖表。
B7 為 Amazon 策略結果與經驗教訓建立交易員審閱紀錄 [Part 3]
把策略排名寫成交易員審閱紀錄。
B8 使用 AgentCore 與 Strands 建立受治理的 FSI Amazon 部位管理 Playbook [Part 4]
受治理的 FSI Amazon 部位管理手冊。
B9 使用 Amazon Bedrock AgentCore 建立主權風險交易代理,分析殖利率差、FX 避險與債務重新定價
主權風險代理:殖利率差、外匯避險與債務重定價。
B11 建置現代波動率交易與合法泰國復原規劃代理程式:記憶體驅動的 Strands 多代理程式風險防護系統
記憶驅動的 Strands 代理:波動率與泰國復原規劃。
B12 使用 Amazon Bedrock AgentCore Memory 建構空頭跨式部位交易風險治理
空頭跨式部位的交易風險治理。
B13 在 Amazon EKS 上建構生產環境就緒的信用與收益質押 AI 智能體
在 EKS 上跑生產級信用與收益質押代理。
挑戰
01 週末生產力挑戰:Fab SPC 漂移同步入口網站
Fab SPC 漂移審查與建議入口網站。
02 週末生產力挑戰:Quant P&L Commander — AWS 上由 AI 驅動的交易生產力入口網站
AWS 上由 AI 驅動的交易生產力入口網站。
03 週末煩人的任務挑戰:交易台在雲端、鏈上、空中執行摘要
DeskPulse 日常交易執行摘要。
04 週末 Agent 挑戰:上午 6 點交易風險審查
無人值守、以證據為基礎的晨間交易風險簡報。
05 Weekend Creative Challenge: Leadership Card Game
瀏覽器版創意引導卡牌。
06 Full Stack Challenge: Community Day Board App
瀏覽器版活動溝通空間。
領導力卡牌
01 Leadership Card Game: 雲端還沒自動化的最後一項本事:像領導者一樣說話
寫給 建構者的現場隨筆——談語言、勇氣,以及 Leadership Card Game
02 領導力回合的解剖:Leadership Card Game 實際怎麼玩
給 建構者的引導員實地指南——讓演練嵌進真實會議
03 Leadership Card Game: 當機會不再屬於主辦者
寫給 建構者的現場隨筆:權力轉移、多語領導力練習夜,以及走完入口、資源、敘事的職涯弧線
04 週末創意挑戰:Leadership Card Game
一篇建構者手記:願景、架構,以及週末創意挑戰教會我的事
05 從週末挑戰專案到 $1,386 群眾募資:改變你在職場現身方式的領導力練習
一個週末做出的作品,變成 600 張卡的線上領導力練習室,並募到 $1,386。
06 從週末挑戰專案到 $1,386 群眾募資:進入科技產業的第一天路徑
一個週末挑戰如何變成具備 600 張卡、由 AWS 驅動的多語產品,並募到 $1,386?
07 從週末挑戰專案到 $1,386 群眾募資:用轉移機會建立專業品牌
一個週末挑戰把領導想法做成能跑的多語產品,並募到 $1,386。
08 Leadership Card Game — 群眾募資活動
募資目標: HKD 5,000 已募金額: HKD 1,386 距目標還差: HKD 3,614 進度: 28% 創作者: D.C. Dan · L.L. Diana · L.K. Lva 所在地: 日本、香港、新加坡 投資人權益: 即期價值、私密會員卡牌編輯器雲(Private Membership Card…
09 PR/FAQ 01 — Leadership Card Game 面向社群建構者正式推出
「逆向工作法」文件 · 對外新聞稿 + FAQ 產品: Leadership Card Game 受眾: 社群經理、志願組織者、職涯早期建構者
10 PR/FAQ 02 — 企業引導員採用 Leadership Card Game 進行現場領導力演練
「逆向工作法」文件 · 對外新聞稿 + FAQ 產品: Leadership Card Game 受眾: 學習與發展負責人、人員管理者、敏捷教練、企業引導員
10 PR/FAQ 03 — 多語 Leadership Card Game 為建構者擁有權開放全球練習室
「逆向工作法」文件 · 對外新聞稿 + FAQ 產品: Leadership Card Game 受眾: 全球 建構者、雙語社群、跨境產品團隊、開源導師
AWS Builder Center
01 AWS Builder Center、社群精神與 AWS Builder Jacket
霓虹訊號、共享創意,以及為建構者打造的外套。
02 走進 AWS Builder Center:一座能學習、貢獻,也讓人有歸屬感的全球技術平台
一段精彩旅程,不一定從機場開始。
03 AWS Community Builder 的巨大成功
當建構者公開分享,整個社群就會一起前進。
04 AWS Builder Center 的巨大成功
一座為好奇心打造、充滿活力的全球街區。
05 週末走進 AWS Builder Center:從社群靈感到令人難忘的 AWS Builder Jacket
星期五晚上,開始於建構者熟悉的感覺:有一個點子,正卡在問題與可能性之間。

對象: 後端開發人員、AI 平台開發人員、AWS 整合工程師

時長: 2 小時

主要 AWS AI 服務: Amazon Bedrock AgentCore Gateway、Strands Agents、MCP、AWS Lambda

專案產出: 具備本地端 MCP 工具、Lambda 支援的 Gateway 目標、語義搜尋以及 Strands 代理呼叫 functional 的託管型 MCP 工具織網 (Tool Fabric)。

本工作坊僅供教育工程用途。範例中所使用的主權風險(sovereign-risk)領域僅用於教學代理工具架構,不構成任何財務建議。


工作坊摘要

開發人員將建構一個託管型的 Gateway MCP 工具織網,用以連接本地端 MCP 原型設計、Lambda 支援的工具、語義搜尋以及 Strands 代理消費(consumption)。本工作坊將引導完成 Schema 設計、本地端探索測試、雲端目標註冊、JSON-RPC 除錯以及 SigV4 傳輸整合。各團隊最終將建立一套可擴展的工具治理模式,在複雜的多團隊平台工程環境中,於 AWS 上實現可探索、安全且導向證據的代理工具。


1. 開發者學習目標

開發人員將學習如何:

● 設計面向代理的工具名稱與 Schema。

● 建構本地端可串流的 HTTP MCP 工具伺服器。

● 在雲端部署前測試 MCP 工具探索。

● 實作由 Lambda 支援的工具商務邏輯。

● 在 Amazon Bedrock AgentCore Gateway 中註冊 Lambda 工具。

● 啟用意圖導向的工具語義搜尋。

● 使用 SigV4 傳輸從 Strands 代理呼叫 Gateway 工具。

● 新增直接的 JSON-RPC 工具呼叫測試以進行除錯。


2. 本工作坊建構之架構

本地端開發環境
  ├─ mcp_server.py               # 本地端 MCP 伺服器
  ├─ mcp_client_local.py         # 本地端探索用戶端
  ├─ lambda_function.py          # Lambda 目標邏輯
  ├─ package_lambda.sh           # Lambda zip 包裝腳本
  ├─ gateway_setup.py            # Gateway 與目標建立
  ├─ semantic_search.py          # Gateway 語義搜尋呼叫
  ├─ direct_tool_call.py         # JSON-RPC tools/call 除錯
  ├─ strands_gateway_agent.py    # 使用 Gateway MCP 工具的 Strands 代理
  └─ prompts/tool_selection.md

AWS
  ├─ Amazon Bedrock AgentCore Gateway
  ├─ AWS Lambda 目標
  ├─ Cognito / 自訂 JWT 授權器輸入
  ├─ IAM Gateway 角色
  └─ 具備 SigV4 MCP 傳輸的 Strands 代理用戶端

3. 2 小時實作議程

時間 (分鐘) 模組 實作產出
0–10 工具織網概念 理解 MCP、Gateway、Lambda 目標流程
10–25 工具契約 規劃工具名稱、描述與 Schema
25–45 本地端 MCP 伺服器 執行可串流的 HTTP 工具伺服器
45–60 本地端探索 驗證工具列表與中介資料 (Metadata)
60–80 Lambda 目標 包裝並部署或準備工具後端
80–100 Gateway 目標 建立 MCP Gateway 與 Lambda 目標
100–110 語義搜尋 搜尋工具並回傳相關工具
110–120 Strands 代理 代理呼叫 Gateway 工具

步驟 1 — 定義工具選擇 Prompt 與 Schema 策略

開發者行動

mkdir -p agentcore-strands-gateway/prompts
cd agentcore-strands-gateway
cat > prompts/tool_selection.md <<'EOF'
您是主權風險工作流程的工具選擇規劃器。
當工具庫規模龐大時,請使用語義搜尋。
選擇能夠回答使用者請求的最精簡工具組合。
回傳選定的工具名稱、引數、預期證據以及選擇理由。
請勿呼叫執行或交易工具。本系統僅用於工程分析。
EOF

建立 tool_schema.py。

TOOL_SCHEMA = [
    {
        "name": "funding_liquidity",
        "description": "評估附買回壓力、貨幣市場壓力、美元融資及現金偏好。",
        "inputSchema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "資融通流動性分析請求。"}},
            "required": ["query"],
        },
    },
    {
        "name": "credit_spread_widening",
        "description": "評估投資級/高收益級 (IG/HY) 利差擴大、CDS 壓力、降評風險及流動性溢價。",
        "inputSchema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "信用利差分析請求。"}},
            "required": ["query"],
        },
    },
    {
        "name": "sovereign_debt_risk_repricing",
        "description": "評估財政可信度、實質殖利率、政策分歧、資本流動及外匯壓力。",
        "inputSchema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "主權風險重估分析請求。"}},
            "required": ["query"],
        },
    },
    {
        "name": "currency_mismatch",
        "description": "評估外匯錯配、外部債務壓力、外匯存底、基差交換及避險背景。",
        "inputSchema": {
            "type": "object",
            "properties": {"query": {"type": "string", "description": "貨幣錯配分析請求。"}},
            "required": ["query"],
        },
    },
]

商務邏輯

Schema 定義了向代理公開的工具詞彙表,同時也定義了每個工具負責處理的使用者意圖。

程式碼邏輯

每個 Schema 項目包含工具名稱、描述、JSON 輸入 Schema 以及必要欄位。Gateway 設定將會重用相同的 Schema。

預期結果

開發人員擁有單一且權威的工具 Schema 來源,無需在多個腳本中重複複製 Schema。

系統設計決策

● 權威 Schema 來源: 工具 Schema 會影響探索、驗證和代理行為。將它們保留在 tool_schema.py 中可防止本地端測試、Gateway 註冊與文件之間產生偏離。隨著工具型錄的增長,這一點至關重要。

● 針對搜尋優化的描述: 描述中包含附買回壓力、CDS 壓力、實質殖利率和外匯壓力等商業術語。語義搜尋極度依賴有意義的中介資料,因此應將描述視為生產級別的 API 文件。

● 最精簡工具組合原則: Prompt 指導代理選擇足夠的最精簡工具組合。這能減少不必要的工具呼叫、降低延遲,並使稽核軌跡更容易審查。


步驟 2 — 建構本地端 MCP 伺服器

開發者行動

建立 mcp_server.py。

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(host="0.0.0.0", port=8000, stateless_http=True)

@mcp.tool()
def funding_liquidity(query: str) -> dict:
    """評估附買回壓力、貨幣市場壓力、美元融資及現金偏好。"""
    return {
        "tool": "funding_liquidity",
        "query": query,
        "signals": ["附買回利率", "交換利差", "美元融資", "現金偏好"],
        "summary": "在將利差變動解讀為純粹的信用風險之前,應先檢查資融通壓力。",
    }

@mcp.tool()
def credit_spread_widening(query: str) -> dict:
    """評估投資級/高收益級 (IG/HY) 利差擴大、CDS 壓力、降評風險及流動性溢價。"""
    return {
        "tool": "credit_spread_widening",
        "query": query,
        "signals": ["IG 利差", "HY 利差", "CDS 指數", "降評觀察"],
        "summary": "信用利差擴大可能反映了違約風險重估與流動性撤出。",
    }

@mcp.tool()
def sovereign_debt_risk_repricing(query: str) -> dict:
    """評估財政可信度、實質殖利率、政策分歧、資本流動及外匯壓力。"""
    return {
        "tool": "sovereign_debt_risk_repricing",
        "query": query,
        "signals": ["實質殖利率", "財政路徑", "央行反應", "資本流動"],
        "summary": "主權風險重估應區分財政風險、政策分歧與資本流動壓力。",
    }

@mcp.tool()
def currency_mismatch(query: str) -> dict:
    """評估外匯錯配、外部債務壓力、外匯存底、基差交換及避險背景。"""
    return {
        "tool": "currency_mismatch",
        "query": query,
        "signals": ["外匯存底", "外部債務", "基差交換", "遠期點數"],
        "summary": "當外部融資收緊時,貨幣錯配可能會放大主權壓力。",
    }

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

執行:

python mcp_server.py

商務邏輯

每個 MCP 工具都會回傳關於特定風險維度的結構化證據。

程式碼邏輯

FastMCP 透過可串流的 HTTP 將 Python 函數公開為工具。每個函數接受一個 query 並回傳一個字典。

預期結果

本地端 MCP 端點可在 http://localhost:8000/mcp 取得。

系統設計決策

● 結構化工具輸出: 回傳字典而非純文字字串,可為代理和測試提供可檢查的欄位:工具名稱、查詢、訊號和摘要。這改善了事實接地(grounding)與未來的評估。

● 建構 Gateway 前先建立本地端伺服器: 本地端 MCP 測試可降低雲端複雜性。開發人員先驗證工具行為和中介資料,然後再將工具發布到 AgentCore Gateway 後方。

● 無狀態工具設計: 工具不依賴隱藏的伺服器記憶體。這使得它們更容易在託管基礎設施後方進行擴展、測試與部署。


步驟 3 — 測試本地端 MCP 探索與直接呼叫

開發者行動

建立 mcp_client_local.py。

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()
            tools = await session.list_tools()
            print("已探索的工具:")
            for tool in tools.tools:
                print("-", tool.name, "|", tool.description)

            result = await session.call_tool(
                "sovereign_debt_risk_repricing",
                {"query": "美中殖利率利差擴大與人民幣壓力"},
            )
            print("直接呼叫結果:")
            print(result)

if __name__ == "__main__":
    asyncio.run(main())

商務邏輯

用戶端在引入 Gateway 之前,先驗證探索功能與直接執行功能。

程式碼邏輯

用戶端初始化一個 MCP 工作階段(Session),列出所有工具,然後傳入引數呼叫其中一個工具。

預期結果

主控台印出工具中介資料與結構化的工具執行結果。

系統設計決策

● 探索加上直接呼叫測試: 單純列出工具只能證明中介資料的可見性。直接呼叫能證明工具可接受引數並回傳有效的回應。在發布至雲端前,兩者皆為必要測試。

● 無模型參與(No model in the loop): 測試將 MCP 傳輸和工具行為與 LLM 推理隔離。這使得中斷與錯誤更容易除錯。

● 具代表性的查詢: 直接呼叫使用包含殖利率利差與人民幣壓力的現實情境查詢。使用網域專屬語言進行測試有助於驗證描述和輸出的實用性。


步驟 4 — 實作 Lambda 目標邏輯

開發者行動

建立 lambda_function.py。

import json

HANDLERS = {
    "funding_liquidity": {
        "signals": ["附買回壓力", "貨幣市場利差", "美元融資", "交換基差"],
        "summary": "在將殖利率變動視為孤立的主權風險重估之前,請先檢查資融通壓力。",
    },
    "credit_spread_widening": {
        "signals": ["IG 利差", "HY 利差", "CDS 指數", "降評風險"],
        "summary": "信用利差擴大可能代表違約風險重估或流動性溢價擴張。",
    },
    "sovereign_debt_risk_repricing": {
        "signals": ["實質殖利率", "財政可信度", "政策分歧", "資本流動"],
        "summary": "主權風險重估應分解為財政、政策、資金流動與外匯驅動因素。",
    },
    "currency_mismatch": {
        "signals": ["外部債務", "外匯存底", "基差交換", "遠期避險成本"],
        "summary": "當外部再融資條件收緊時,貨幣錯配可能會放大壓力。",
    },
}

def lambda_handler(event, context):
    tool_name = event.get("toolName") or event.get("name")
    arguments = event.get("arguments", {})
    query = arguments.get("query", "")

    if tool_name not in HANDLERS:
        return {
            "statusCode": 404,
            "body": json.dumps({"error": f"未知的工具:{tool_name}"}),
        }

    output = {
        "tool": tool_name,
        "query": query,
        "signals": HANDLERS[tool_name]["signals"],
        "summary": HANDLERS[tool_name]["summary"],
    }
    return {"statusCode": 200, "body": json.dumps(output)}

建立封裝輔助腳本:

cat > package_lambda.sh <<'EOF'
#!/usr/bin/env bash
set -euo pipefail
rm -f lambda_function.zip
zip lambda_function.zip lambda_function.py
ls -lh lambda_function.zip
EOF
chmod +x package_lambda.sh
./package_lambda.sh

商務邏輯

Lambda 成為受監管的 Gateway 工具呼叫後端。它將工具請求對映到結構化的訊號輸出。

程式碼邏輯

處理常式(Handler)讀取 toolName 和 arguments,驗證工具,並回傳 JSON。封裝腳本則建立一個 zip 成品。

預期結果

lambda_function.zip 已準備就緒,可用於部署或用於預先準備的工作坊引導腳本。

系統設計決策

● Lambda 作為後端邊界: Gateway 負責公開工具,而 Lambda 擁有確定性的商務邏輯。這將模型推理與後端執行分離,使工具行為具備可測試性與可觀測性。

● 共享處理器對映: 基於字典 carbon 路由機制在保持工作坊精簡的同時,展示了多個工具如何對映到單一 Lambda 目標。生產系統隨後可將處理常式拆分。

● 結構化回應內文: 回傳 tool、query、signals 和 summary,為代理提供了在綜合分析前可引用的證據。這也提高了可稽核性。


步驟 5 — 建立 AgentCore Gateway 並註冊目標

開發者行動

建立 gateway_setup.py。

import os
import boto3
from tool_schema import TOOL_SCHEMA

region = os.getenv("AWS_DEFAULT_REGION", "us-east-1")
client = boto3.client("bedrock-agentcore-control", region_name=region)

gateway = client.create_gateway(
    name="gateway-sovereign-risk-tools-hands-on",
    roleArn=os.environ["AGENTCORE_GATEWAY_ROLE_ARN"],
    protocolType="MCP",
    authorizerType="CUSTOM_JWT",
    authorizerConfiguration={
        "customJWTAuthorizer": {
            "allowedClients": [os.environ["COGNITO_CLIENT_ID"]],
            "discoveryUrl": os.environ["COGNITO_DISCOVERY_URL"],
        }
    },
    protocolConfiguration={
        "mcp": {
            "searchType": "SEMANTIC",
            "supportedVersions": ["2025-03-26"],
        }
    },
    description="主權風險工具實作 MCP Gateway",
)

target = client.create_gateway_target(
    gatewayIdentifier=gateway["gatewayId"],
    name="sovereign-risk-lambda-target",
    description="公開主權風險工具的 Lambda 目標",
    targetConfiguration={
        "mcp": {
            "lambda": {
                "lambdaArn": os.environ["SOVEREIGN_TOOLS_LAMBDA_ARN"],
                "toolSchema": {"inlinePayload": TOOL_SCHEMA},
            }
        }
    },
    credentialProviderConfigurations=[{"credentialProviderType": "GATEWAY_IAM_ROLE"}],
)

print("Gateway ID:", gateway["gatewayId"])
print("Gateway URL:", gateway["gatewayUrl"])
print("Target ID:", target["targetId"])

商務邏輯

Gateway 透過託管的 MCP 端點發布由 Lambda 支援的主權風險工具。

程式碼邏輯

此腳本建立 Gateway、設定 JWT 授權、啟用語義搜尋,並使用權威 Schema 註冊 Lambda 目標。

預期結果

開發人員獲得 Gateway URL 和目標 ID。

系統設計決策

● Gateway 集中化工具存取: AgentCore Gateway 成為代理的託管 MCP 端點。這避免了每個代理都必須直接與 Lambda、驗證和傳輸邏輯整合。

● JWT 入口與 IAM 出口: 入站呼叫者使用 JWT 進行驗證,而 Gateway 使用 IAM 叫用 Lambda。這將呼叫者授權與後端執行憑證分離。

● 啟用語義探索: Gateway 對工具中介資料進行索引以供語義搜尋使用。這為大型工具型錄做好準備,避免因將所有 Schema 塞入代理內容(Context)而導致 Prompt 膨脹。


步驟 6 — 測試 Gateway 語義搜尋與直接 JSON-RPC 呼叫

開發者行動

建立 semantic_search.py。

import json
import os
import requests

def call_gateway(payload):
    response = requests.post(
        os.environ["AGENTCORE_GATEWAY_URL"],
        json=payload,
        headers={
            "Authorization": f"Bearer {os.environ['AGENTCORE_GATEWAY_JWT']}",
            "Content-Type": "application/json",
        },
        timeout=30,
    )
    response.raise_for_status()
    return response.json()

payload = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
        "name": "x_amz_bedrock_agentcore_search",
        "arguments": {"query": "人民幣壓力與主權債務風險重估"},
    },
}

print(json.dumps(call_gateway(payload), indent=2))

建立 direct_tool_call.py。

import json
import os
import requests

payload = {
    "jsonrpc": "2.0",
    "id": 2,
    "method": "tools/call",
    "params": {
        "name": "sovereign_debt_risk_repricing",
        "arguments": {"query": "美中殖利率利差擴大與資本流動壓力"},
    },
}

response = requests.post(
    os.environ["AGENTCORE_GATEWAY_URL"],
    json=payload,
    headers={
        "Authorization": f"Bearer {os.environ['AGENTCORE_GATEWAY_JWT']}",
        "Content-Type": "application/json",
    },
    timeout=30,
)
print(json.dumps(response.json(), indent=2))

商務邏輯

語義搜尋透過意圖尋找相關工具。直接呼叫則驗證選定的工具是否能透過 Gateway 正確執行。

程式碼邏輯

這兩個腳本都發送 JSON-RPC tools/call 請求。一個呼叫內建的搜尋工具;另一個呼叫定義的網域工具。

預期結果

搜尋回傳排名後的工具,直接呼叫回傳由 Lambda 支援的結構化證據。

系統設計決策

● 搜尋與直接執行為獨立測試: 語義搜尋驗證探索功能。直接呼叫驗證執行功能。保持兩者獨立有助於隔離驗證、Schema、路由或 Lambda 邏輯中的錯誤。

● JSON-RPC 除錯路徑: 當代理行為異常時,直接 HTTP 呼叫非常有用。開發人員可以重現精確的 Gateway 請求,而無需將模型推理引入流程。

● Bearer 權杖邊界: Gateway 呼叫包含授權標頭(Authorization Headers)。這強化了工具探索與執行是受保護的操作,而非匿名 API。


步驟 7 — 從 Strands 代理呼叫 Gateway

開發者行動

建立 strands_gateway_agent.py。

import os
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 streamable_http_sigv4 import streamablehttp_client_with_sigv4

SERVICE = "bedrock-agentcore"

def assume_gateway_role(role_arn: str):
    return boto3.client("sts").assume_role(
        RoleArn=role_arn,
        RoleSessionName="strands-gateway-agent-hands-on",
        DurationSeconds=3600,
    )["Credentials"]

def main():
    region = os.getenv("AWS_DEFAULT_REGION", "us-east-1")
    creds = assume_gateway_role(os.environ["AGENTCORE_GATEWAY_INVOKE_ROLE_ARN"])

    mcp_client = MCPClient(lambda: streamablehttp_client_with_sigv4(
        url=os.environ["AGENTCORE_GATEWAY_URL"],
        credentials=Credentials(
            creds["AccessKeyId"],
            creds["SecretAccessKey"],
            creds["SessionToken"],
        ),
        service=SERVICE,
        region=region,
    ))

    with mcp_client:
        tools = mcp_client.list_tools_sync()
        print("已載入的工具:", [tool.tool_name for tool in tools])

        agent = Agent(
            model=BedrockModel(model_id="amazon.nova-pro-v1:0", temperature=0.2),
            tools=tools,
            system_prompt=(
                "您是使用工具的風險工程助理。 "
                "請先使用 Gateway 工具取得證據。接著綜合分析限制、確認訊號及失效觸發條件。 "
                "請勿提供投資建議或自主交易指令。"
            ),
        )

        result = agent(
            "分析美中殖利率利差擴大所帶來的人民幣壓力、主權債務風險重估以及信用利差擴大。"
        )
        print(result)

if __name__ == "__main__":
    main()

商務邏輯

Strands 代理在撰寫最終綜合分析之前,將 Gateway 工具用作證據來源。

程式碼邏輯

該腳本扮演一個 IAM 角色、建立 SigV4 MCP 傳輸、列出 Gateway 工具、將它們載入到 Strands 中並叫用代理。

預期結果

代理印出已載入的工具,並回傳基於工具證據的分析。

系統設計決策

● Strands 使用託管型 MCP 工具: 代理不直接呼叫 Lambda。Gateway 擁有工具公開、驗證和轉換的所有權。這使得代理獨立於後端實作細節。

● 暫時性憑證: 代理使用假設角色(Assumed-role)憑證,減少了長期金鑰的外洩風險。IAM 定義了代理擁有哪些 Gateway 存取權限。

● 證據優先的代理 Prompt: 系統 Prompt 要求在綜合分析前先取得工具證據。這提高了可稽核性,並使輸出更容易評估。


開發者最終檢查清單

● [ ] 本地端 MCP 伺服器正常執行。

● [ ] 本地端 MCP 用戶端能列出並呼叫工具。

● [ ] Lambda zip 封裝檔案已存在。

● [ ] Gateway 與目標已建立。

● [ ] 語義搜尋能回傳相關工具。

● [ ] 直接 JSON-RPC 工具呼叫成功。

● [ ] Strands 代理成功載入 Gateway 工具並產生有根據的輸出。


額外實作開發者實驗室

這些實驗室擴展了 Gateway MCP 工具織網工作坊,涵蓋更深入的除錯、Schema 治理、驗證實務以及代理工具評估。它們刻意與核心建構流程不同,專注於讓工具生態系統達到生產就緒狀態。


實作實驗室 A — 為 MCP 工具定義新增 Schema 檢查 (Linting)

開發者目標

在向 AgentCore Gateway 註冊之前,先驗證工具 Schema 的正確性與品質。

開發者行動

建立 lint_tool_schema.py:

from tool_schema import TOOL_SCHEMA

REQUIRED_TOP_LEVEL = {"name", "description", "inputSchema"}

def lint_tool_schema(schema: list[dict]) -> list[str]:
    errors = []
    names = set()
    for index, tool in enumerate(schema):
        missing = REQUIRED_TOP_LEVEL - set(tool.keys())
        if missing:
            errors.append(f"工具索引 {index} 缺少鍵值:{sorted(missing)}")

        name = tool.get("name")
        if not name:
            errors.append(f"工具索引 {index} 的名稱為空")
        elif name in names:
            errors.append(f"重複的工具名稱:{name}")
        names.add(name)

        description = tool.get("description", "")
        if len(description.split()) < 4:
            errors.append(f"工具 {name} 的描述太短,不利於語義搜尋")

        input_schema = tool.get("inputSchema", {})
        if input_schema.get("type") != "object":
            errors.append(f"工具 {name} 的 inputSchema 必須是 object")
        if "required" not in input_schema:
            errors.append(f"工具 {name} 的 inputSchema 應定義必要欄位 (required)")
    return errors

if __name__ == "__main__":
    errors = lint_tool_schema(TOOL_SCHEMA)
    if errors:
        print("Schema 檢查失敗")
        print("\n".join(errors))
        raise SystemExit(1)
    print(f"已成功驗證 {len(TOOL_SCHEMA)} 個工具 Schema")

執行:

python lint_tool_schema.py

商務邏輯

工具 Schema 是代理 API 的一部分。不良的 Schema 會降低探索品質,並可能導致執行階段的叫用失敗。

程式碼邏輯

此程式碼檢查器(Linter)會檢查必要欄位、重複的工具名稱、描述品質、物件 Schema 以及必要的輸入欄位。

預期結果

已成功驗證 4 個工具 Schema

系統設計決策

● 註冊前把關 Schema 品質: Gateway 註冊不應該是第一次檢查 Schema 的地方。本地端 Linting 可在雲端資源變更之前捕捉到簡單的問題,例如重複名稱和不夠具體的描述。

● 語義搜尋依賴描述: 簡短或模糊的描述會降低搜尋索引的實用性。Linter 強制執行最低描述品質,因為中介資料對代理而言就是運作行為。

● CI 易整合的快速檢查: 當驗證失敗時,腳本會以狀態碼 1 結束,使其可以輕鬆地在 CI 流水線或 pre-commit 勾子(Hooks)中執行。


實作實驗室 B — 建構直接面向 Gateway 的 MCP tools/list 測試器

開發者目標

在涉及 Strands 之前,使用原始的 JSON-RPC 偵錯 Gateway 工具探索功能。

開發者行動

建立 gateway_list_tools.py:

import json
import os
import requests

payload = {
    "jsonrpc": "2.0",
    "id": 100,
    "method": "tools/list",
    "params": {},
}

response = requests.post(
    os.environ["AGENTCORE_GATEWAY_URL"],
    json=payload,
    headers={
        "Authorization": f"Bearer {os.environ['AGENTCORE_GATEWAY_JWT']}",
        "Content-Type": "application/json",
    },
    timeout=30,
)

print("HTTP 狀態碼:", response.status_code)
print(json.dumps(response.json(), indent=2))

商務邏輯

在代理能夠呼叫工具之前,工具探索必須先正常工作。本實驗室直接驗證 Gateway 的探索功能。

程式碼邏輯

該腳本向帶有 Bearer 權杖的 Gateway 端點發送 JSON-RPC tools/list 請求。

預期結果

回應包含所有已註冊的 MCP 工具及其 Schema。

系統設計決策

● 原始協定除錯: 當 Strands 代理行為令人困惑時,開發人員需要一個更低階的測試。JSON-RPC 呼叫將 Gateway 探索功能與模型推理及 Strands 編排隔離。

● 驗證可見性: 該腳本使 Bearer 權杖要求變得明確。這有助於開發人員區分是驗證失敗還是工具註冊失敗。

● 維運冒煙測試: tools/list 可以成為部署冒煙測試(Smoke Test),以確認新註冊的 Gateway 目標在註冊後是否可見。


實作實驗室 C — 新增 Gateway 工具呼叫重放檔案 (Replay File)

開發者目標

建立可重複使用的 JSON-RPC 測試負載,以便在除錯期間進行重放。

開發者行動

建立 requests/sovereign_repricing_call.json:

{
  "jsonrpc": "2.0",
  "id": 201,
  "method": "tools/call",
  "params": {
    "name": "sovereign_debt_risk_repricing",
    "arguments": {
      "query": "美中殖利率利差擴大、人民幣壓力與資本流動壓力"
    }
  }
}

建立 replay_gateway_request.py:

import json
import os
import sys
import requests

path = sys.argv[1]
payload = json.load(open(path, encoding="utf-8"))

response = requests.post(
    os.environ["AGENTCORE_GATEWAY_URL"],
    json=payload,
    headers={
        "Authorization": f"Bearer {os.environ['AGENTCORE_GATEWAY_JWT']}",
        "Content-Type": "application/json",
    },
    timeout=30,
)
print(json.dumps(response.json(), indent=2))

執行:

mkdir -p requests
python replay_gateway_request.py requests/sovereign_repricing_call.json

商務邏輯

重放檔案為除錯工具行為和展示預期的 Gateway 回應建立了可重複的證據。

程式碼邏輯

重放腳本從磁碟載入 JSON 負載並將其張貼(Post)到 Gateway。

預期結果

相同的工具呼叫可以重複重放,而無需重寫 Python 程式碼。

系統設計決策

● 可重現的工具除錯: 儲存的 JSON-RPC 負載使得重現 Bug 和比較跨部署的回應變得容易。當工具 Schema 或 Lambda 邏輯變更時,這非常有用。

● 請求與執行器的分離: 請求檔案是資料,而 Python 腳本是通用的執行器。這允許許多測試案例重用同一個用戶端。

● 具備文件價值: 重放檔案可兼作範例,供其他需要了解如何直接呼叫 Gateway 工具的開發人員參考。


實作實驗室 D — 為 Strands 回應新增工具結果標準化

開發者目標

在代理綜合最終答案之前,先將工具輸出標準化(Normalize)。

開發者行動

建立 tool_result_normalizer.py:

import json

def normalize_tool_result(raw_result) -> dict:
    if isinstance(raw_result, dict):
        return raw_result

    if isinstance(raw_result, str):
        try:
            return json.loads(raw_result)
        except json.JSONDecodeError:
            return {"raw_text": raw_result}

    if isinstance(raw_result, list):
        return {"items": raw_result}

    return {"repr": repr(raw_result)}

在除錯程式碼中使用它:

from tool_result_normalizer import normalize_tool_result

# 假設 result 是來自先前工具呼叫的輸出
normalized = normalize_tool_result(result)
print(normalized)

商務邏輯

代理可能會收到各種不同形狀的工具輸出。標準化使得下游的證據處理變得可預測。

程式碼邏輯

此輔助工具將字典、JSON 字串、列表和未知物件轉換為統一的字典形狀。

預期結果

工具輸出可以被一致地記錄、評估與呈現。

系統設計決策

● 可預測的證據格式: 工具輸出應盡可能可供機器讀取。標準化允許用戶端和評估機制檢查欄位,而不必依賴每個後端都回傳完全相同的形狀。

● 防禦性整合: Gateway 目標可能會演進或回傳非預期的負載。標準化可防止脆弱的代理包裝器(Wrappers)因微小的輸出形狀變更而失敗。

● 評估準備就緒: 一致的工具證據使得檢查最終代理回應是否使用了預期的工具和訊號變得更加容易。


實作實驗室 E — 新增語義搜尋比較測試

開發者目標

比較不同使用者意圖的語義搜尋結果,並驗證預期工具是否出現在前幾名結果中。

開發者行動

建立 semantic_search_eval.py:

import json
import os
import requests

CASES = [
    {"query": "附買回融資壓力與現金偏好", "expected": "funding_liquidity"},
    {"query": "CDS 擴大與降評風險", "expected": "credit_spread_widening"},
    {"query": "財政可信度與實質殖利率重估", "expected": "sovereign_debt_risk_repricing"},
    {"query": "外匯存底與外部美元債務", "expected": "currency_mismatch"},
]

def search(query: str) -> list[str]:
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": {
            "name": "x_amz_bedrock_agentcore_search",
            "arguments": {"query": query},
        },
    }
    response = requests.post(
        os.environ["AGENTCORE_GATEWAY_URL"],
        json=payload,
        headers={"Authorization": f"Bearer {os.environ['AGENTCORE_GATEWAY_JWT']}"},
        timeout=30,
    ).json()
    tools = response.get("result", {}).get("structuredContent", {}).get("tools", [])
    return [tool.get("name") for tool in tools]

for case in CASES:
    names = search(case["query"])
    passed = case["expected"] in names[:3]
    print(case["query"], "【通過】" if passed else "【失敗】", names[:3])

商務邏輯

語義搜尋的品質決定了代理是否能針對商業語言請求找到正確的工具。

程式碼邏輯

該腳本執行多個搜尋查詢,並檢查預期工具是否出現在前三個結果中。

預期結果

每個測試案例都會印出「通過」或「失敗」以及前三名匹配的工具。

系統設計決策

● 搜尋品質作為可測試行為: 語義搜尋應像任何其他功能一樣進行驗證。如果描述或 Schema 變更,搜尋品質可能會退化。

● 基於意圖的測試案例: 案例使用商業語言而非精確的工具名稱。這測試了中介資料是否能支持真實的使用者 Prompt。

● Top-K 容差: 檢查前三個結果比每次都要求第一個結果更具現實意義。語義檢索對相似工具的排名可能略有不同,但預期的工具仍應保持可探索性。


實作實驗室 F — 新增最低權限環境檢查清單

開發者目標

在晉升到生產環境之前,為 Gateway 工具織網建立一份維運檢查清單。

開發者行動

建立 docs/gateway_operational_checklist.md:

# Gateway 維運檢查清單

## 身分識別與授權
- [ ] Gateway 使用 CUSTOM_JWT 授權器。
- [ ] 允許的用戶端限制在經批准的應用程式用戶端。
- [ ] 權杖(Token)生命週期和重新整理(Refresh)程序已記載於文件。

## IAM 與目標存取
- [ ] Gateway 角色僅能叫用經批准的 Lambda 目標。
- [ ] Lambda 資源政策不允許廣泛的公開叫用。
- [ ] 開發人員憑證未內嵌於程式碼中。

## 工具治理
- [ ] 工具 Schema 在註冊前已通過 Lint 檢查。
- [ ] 工具描述長度足夠,利於語義搜尋。
- [ ] 工具輸出為結構化且具備可稽核性。
- [ ] 危險或執行/交易導向的工具未註冊到此分析 Gateway。

## 可觀測性
- [ ] 已監控 Gateway 目標叫用錯誤。
- [ ] Lambda 日誌包含請求 ID 或工具名稱。
- [ ] Schema 變更後語義搜尋測試均通過。

商務邏輯

檢查清單可幫助平台團隊在廣泛公開工具之前,審查安全性、治理和可觀測性。

程式碼邏輯

這是 Markdown 文件,但它會成為 Pull Request 和生產環境審查的維運構件(Artifact)。

預期結果

團隊擁有一份可用於 Gateway 工具織網的重複性就緒檢查清單。

系統設計決策

● 文件即控制(Documentation as control): 維運檢查清單可防止重要的生產環境疑慮流於部落知識。當多個團隊發布工具時,這特別有用。

● 聚焦最低權限: Gateway 集中了工具存取權,因此必須仔細審查 IAM 和 JWT 邊界。檢查清單使這些邊界變得明確。

● 晉升閘門(Promotion Gate): 檢查清單可以作為一個發布要求,在 Gateway 目標從工作坊移至共享開發或生產環境之前進行核對。