View a markdown version of this page

设置 AWS MCP 服务器 - 亚马逊 OpenSearch 服务

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

设置 AWS MCP 服务器

AWS MCP 服务器是代理工具包中的托管模型上下文协议服务器,用于。 AWS它为您的 AI 代理提供经过身份验证的访问权限 AWS — 代表您运行 AWS API 调用、在沙盒环境中执行脚本以及按需提供 AWS 技能(包括亚马逊 OpenSearch 服务技能)。使用它来处理来自代理的 OpenSearch 服务域和 OpenSearch 无服务器集合。

注意

AWS MCP 服务器连接到公共亚马逊 OpenSearch 服务终端节点。它不支持在 VPC 中部署的域或集合。如果您的 OpenSearch 服务域或 OpenSearch 无服务器集合使用 VPC 访问权限,请改用开源 opensearch-mcp-server-py,它支持通过本地配置进行 VP C 连接。

先决条件

  • 一款支持 Agent Toolkit 的 AI 编码代理 AWS (例如 Kiro、Claude Code、Cursor 和 Codex)。

  • AWS 在本地配置的凭据(例如, AWS CLI 配置文件或环境凭证),允许您执行要运行的 OpenSearch 操作。有关域的信息,请参阅Identity-based 政策;对于无服务器,请参阅Amazon OpenSearch 无服务器的数据访问控制

安装 aws-data-Analytics 插件(推荐)

aws-data-analytics插件将 AWS MCP 服务器配置和amazon-opensearch-service技能捆绑在一次安装中,因此您不必单独配置 MCP 端点或安装技能。

克劳德密码

/plugin install aws-data-analytics@claude-plugins-official /reload-plugins

手抄本

codex plugin marketplace add aws/agent-toolkit-for-aws

然后启动 Codex 并运行浏览/plugins并安装该aws-data-analytics插件。

配置 AWS 直接使用 MCP 服务器

如果您的代理不支持插件,请直接配置 AWS MCP 服务器。在代理工具包中设置 AWS MCP 服务器以获取 AWS 文档,然后向您的代理提 OpenSearch 问 — 它会在运行时通过服务器发现amazon-opensearch-service技能。

使用 AWS 没有技能的 MCP 服务器

无需加载amazon-opensearch-service技能即可直接使用 AWS MCP 服务器。在此模式下,代理使用服务器的内置call_aws工具直接调用 AWS API 操作。这对于临时操作或想要精确控制代理进行的 API 调用时非常有用。

例如,要创建 OpenSearch 无服务器集合,请执行以下操作:

Create an OpenSearch Serverless search collection named "my-collection" in us-east-1

代理使用您配置的凭据代表您调用 OpenSearch 无服务器 CreateCollection API:

{ "name": "my-collection", "type": "SEARCH", "description": "My search collection" }

您也可以在没有技能的情况下对现有资源执行操作。例如:

List all my OpenSearch Service domains in us-east-1 Show the cluster health for my domain named production-domain Create an index called products in my collection endpoint

AWS MCP 服务器处理身份验证,并将您的自然语言请求转换为相应的 AWS API 或 OpenSearch API 调用。

验证设置

安装后,重新启动您的代理,使其加载新的配置。然后问:

What AWS skills do you have available for OpenSearch?

代理应报告amazon-opensearch-service技能。尝试诸如 “列出我的 OpenSearch 服务域” 之类的任务 — 代理使用您配置的凭据通过 AWS MCP 服务器运行呼叫。

安全注意事项

AWS MCP 服务器使用您提供的凭据运行。请遵循以下做法:

  • 使用权限最低的凭证。将专用 IAM 委托人的范围限定为代理所需的 OpenSearch 资源和操作。避免使用管理员凭据。

  • 将开发和生产分开。将服务器指向非生产资源进行探索,并要求在生产变更之前进行确认。

  • 保护凭证。优先选择 IAM 角色和 AWS CLI 配置文件而不是静态访问密钥,并且切勿在源代码管理中向配置文件提交机密。

  • 查看工具输出。MCP 工具响应作为上下文返回到模型。避免对包含不想向模型提供者公开的敏感数据的索引运行服务器。

问题排查

  • 代理看不到任何 OpenSearch 技能或工具 — 确认插件或 MCP 配置有效并完全重启您的代理。大多数代理仅在启动时加载 MCP 服务器。

  • 调用返回 403 禁止 — 您的凭据缺少调用 API 的权限。对于域名,请查看您的委托人的域访问策略和 IAM 策略。对于无服务器,请查看集合的数据访问策略。

与代理框架一起使用

您可以将 AWS MCP 服务器集成到用于编程工作流程的 Python 代理框架中。以下示例连接到 AWS MCP 服务器端点,该端点为 OpenSearch 和其他 AWS 服务提供经过身份验证的访问权限。

Strands Agents

Strands Agents 是一款 AWS原生代理 SDK,内置 MCP 支持,Amazon Bedrock 是默认模型提供商。

from strands import Agent from strands.tools.mcp import MCPClient from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client ENDPOINT = "https://aws-mcp.us-east-1.api.aws/mcp" AWS_REGION = "us-east-1" opensearch_client = MCPClient( lambda: aws_iam_streamablehttp_client( endpoint=ENDPOINT, aws_service="aws-mcp", aws_region=AWS_REGION, ) ) with opensearch_client: agent = Agent(tools=opensearch_client.list_tools_sync()) response = agent("What OpenSearch Service domains do I have?") print(response)

安装所需的软件包:

pip install strands-agents mcp-proxy-for-aws

LangGraph

LangGraph是用于构建有状态代理的低级编排框架。以下示例用于mcp-proxy-for-aws提供经过身份验证的传输,并将 AWS MCP 工具加载langchain-mcp-adapters到由 Amazon Bedrock 支持的 LangChain 代理中。

import asyncio from mcp import ClientSession from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client from langchain_mcp_adapters.tools import load_mcp_tools from langchain_aws import ChatBedrock from langchain.agents import create_agent ENDPOINT = "https://aws-mcp.us-east-1.api.aws/mcp" AWS_REGION = "us-east-1" async def main(): async with aws_iam_streamablehttp_client( endpoint=ENDPOINT, aws_service="aws-mcp", aws_region=AWS_REGION, ) as (read_stream, write_stream, _get_session_id): async with ClientSession(read_stream, write_stream) as session: await session.initialize() tools = await load_mcp_tools(session) model = ChatBedrock( model_id="us.anthropic.claude-opus-4-8", region_name=AWS_REGION, ) agent = create_agent(model, tools) result = await agent.ainvoke( {"messages": [{"role": "user", "content": "What OpenSearch Service domains do I have?"}]} ) print(result["messages"][-1].content) asyncio.run(main())

安装所需的软件包:

pip install mcp-proxy-for-aws langchain langchain-aws langchain-mcp-adapters langgraph

更多信息