Workshop Series
- 工作坊 1:使用 AgentCore 與 Strands 建構:Gateway MCP 工具織網開發者工作坊
- 工作坊 2:使用 AgentCore 與 Strands 建構:受治理的多 Agent 風險系統開發者工作坊
- 工作坊 3:使用 AgentCore 與 Strands 建構:執行期主權風險代理人開發者工作坊
使用 AgentCore 與 Strands 建構:Gateway MCP 工具織網開發者工作坊
對象: 後端開發人員、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 目標從工作坊移至共享開發或生產環境之前進行核對。
