实用提示：使用 Amazon Bedrock AgentCore 对 MCP 工具进行语义搜索

本指南将展示如何从原始材料构建出一个可运行的系统，实现通过 Amazon Bedrock AgentCore Gateway 对 MCP 工具进行语义搜索的功能。重点在于可操作的步骤、明确的检查点，以及可直接放入代码库的代码，无需猜测其用途。 在修改代码之前，应先明确输入参数、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需推测隐藏的状态。 可将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功标准，避免出现无声的、不完整的完成情况。

你将构建什么

在编写“你所构建的内容”时，首先需明确相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外的费用支出。 需为每次调用记录工具名称、参数哈希值、延迟时间以及最终结果。若没有这些记录，调试循环将会耗费大量时间。

为何要在网关端实现语义搜索

在研究 Gateway 端的语义搜索功能时，首先需明确相关规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 应将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中管理，这样操作人员无需查看整个系统结构即可进行审计。 需为每次调用记录工具名称、参数哈希值、延迟时间以及执行结果。若没有这些记录，调试代理循环问题将会耗费大量时间。

前提条件

在处理前置条件时，首先需明确接口规范：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及错误消息处理都是产品功能的一部分，而非后续需要补充的内容。 需为每次调用记录工具名称、参数哈希值、响应延迟及最终结果。没有这些记录，调试过程将会浪费大量时间。 在处理前置条件时，首先需明确接口规范：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关产物命名，定义成功判定标准，杜绝无声的半完成状态。

执行语义搜索

将语义搜索视为可度量的对象来处理时，其效果最佳。在扩大应用范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。在功能结果旁还需记录处理时间以及令牌或查询成本。提前了解这些成本信息，可避免在系统从演示环境过渡到共享环境时出现意外费用。

第一步：查询搜索工具

第一步：将搜索工具视为可度量的对象来查询最为有效。在扩大范围之前，先记录一个成功的案例、一个失败的案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放，以便操作人员无需查看整个系统结构即可进行审计。 使用结构清晰且带有明确副作用标签的工具。主机需要在自动批准之前知道哪些调用会改变系统状态。

from mcp.client.streamable_http import streamablehttp_client
from mcp.client.session import ClientSession

async with streamablehttp_client(gateway_url) as (r, w, _):
    async with ClientSession(r, w) as session:
        await session.initialize()        result = await session.call_tool(
            "x_amz_bedrock_agentcore_search",
            {"query": "find a customer by phone number"},
        )        for match in result.content:
            print(match.text)

第二步：利用这些结果筛选工具列表

第二步：将结果用于筛选工具列表，此时将其视为可测量的对象效果最佳。在扩大范围之前，需记录一个理想运行案例、一个故障案例以及回滚说明。 同时记录正常流程与恢复流程。重试机制、人工审核环节以及错误处理都属于产品本身的功能，而非后续需要补充的内容。 应使用结构清晰且带有明确副作用标签的工具。主机需要在自动批准之前知道哪些调用会改变状态。

async def smart_tool_selection(session, user_request: str, top_k: int = 5):
    search = await session.call_tool(
        "x_amz_bedrock_agentcore_search",
        {"query": user_request},
    )

    relevant_tool_names = [match.text for match in search.content[:top_k]]    all_tools = await session.list_tools()
    return [t for t in all_tools.tools if t.name in relevant_tool_names]

第三步：将其接入 Strands 代理

第三步：将其接入 Strands 代理时，同样需将其视为可测量的对象效果最佳。在扩大范围之前，需记录一个理想运行案例、一个故障案例以及回滚说明。 相比复杂的脚本，应优先选择小型且可测试的单元。当某一步骤出现故障时，故障原因应能明确指向某个特定功能，而非整个复杂的流程。

from strands import Agent
from strands.tools.mcp import MCPClient

async def run(user_message: str):
    async with MCPClient(gateway_url) as mcp:
        relevant_tools = await smart_tool_selection(mcp.session, user_message)        agent = Agent(
            model="anthropic.claude-opus-4-7-v1:0",
            tools=relevant_tools,
            system_prompt="Use only the provided tools to answer.",
        )
        return await agent.run_async(user_message)

搜索参数调整

参考资料

后续内容

操作检查清单