使用 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 小时实施议程

---

步骤 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 目标从工作坊移至共享开发或生产环境之前进行核对。