AI导哈

OpenSearch MCP服务器

一个为OpenSearch提供模型上下文协议（MCP）服务的服务器，通过stdio和SSE服务器暴露4种工具，用于索引列表、映射检索、索引搜索和分片信息获取。

  • by rithinpullela
  • Python
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 rithin-pullela-aws/opensearch-mcp-server

使用说明

注意：此项目已毕业并转移到 opensearch mcp服务器py 存储库。再见！此存储库现在已存档。

OpenSearch MCP服务器

OpenSearch的最小模型上下文协议（MCP）服务器，通过stdio和sse服务器公开4个工具。

可用工具

  • ListIndexTool：列出OpenSearch中的所有索引。
  • IndexMappingTool：检索OpenSearch中索引的索引映射和设置信息。
  • SearchIndexTool：使用OpenSearch中用查询域特定语言（DSL）编写的查询来搜索索引。
  • GetShardsTool:获取OpenSearch中分片的相关信息。

更多工具即将推出。 点击此处

用户指南

安装

从PyPI安装：

pip install test-opensearch-mcp

配置

身份验证方法：

  • 基本身份验证
export OPENSEARCH_URL="<your_opensearch_domain_url>"
export OPENSEARCH_USERNAME="<your_opensearch_domain_username>"
export OPENSEARCH_PASSWORD="<your_opensearch_domain_password>"
  • IAM角色身份验证
export OPENSEARCH_URL="<your_opensearch_domain_url>"
export AWS_REGION="<your_aws_region>"
export AWS_ACCESS_KEY="<your_aws_access_key>"
export AWS_SECRET_ACCESS_KEY="<your_aws_secret_access_key>"
export AWS_SESSION_TOKEN="<your_aws_session_token>"

运行服务器

# Stdio Server
python -m mcp_server_opensearch

# SSE Server
python -m mcp_server_opensearch --transport sse

Claude桌面集成

  • 使用已发布 PyPI包 （推荐）
{
    "mcpServers": {
        "opensearch-mcp-server": {
            "command": "uvx",
            "args": [
                "test-opensearch-mcp"
            ],
            "env": {
                // Required
                "OPENSEARCH_URL": "<your_opensearch_domain_url>",

                // For Basic Authentication
                "OPENSEARCH_USERNAME": "<your_opensearch_domain_username>",
                "OPENSEARCH_PASSWORD": "<your_opensearch_domain_password>",

                // For IAM Role Authentication
                "AWS_REGION": "<your_aws_region>",
                "AWS_ACCESS_KEY": "<your_aws_access_key>",
                "AWS_SECRET_ACCESS_KEY": "<your_aws_secret_access_key>",
                "AWS_SESSION_TOKEN": "<your_aws_session_token>"
            }
        }
    }
}
  • 使用已安装的软件包（通过pip）：
{
    "mcpServers": {
        "opensearch-mcp-server": {
            "command": "python",  // Or full path to python with PyPI package installed
            "args": [
                "-m",
                "mcp_server_opensearch"
            ],
            "env": {
                // Required
                "OPENSEARCH_URL": "<your_opensearch_domain_url>",

                // For Basic Authentication
                "OPENSEARCH_USERNAME": "<your_opensearch_domain_username>",
                "OPENSEARCH_PASSWORD": "<your_opensearch_domain_password>",

                // For IAM Role Authentication
                "AWS_REGION": "<your_aws_region>",
                "AWS_ACCESS_KEY": "<your_aws_access_key>",
                "AWS_SECRET_ACCESS_KEY": "<your_aws_secret_access_key>",
                "AWS_SESSION_TOKEN": "<your_aws_session_token>"
            }
        }
    }
}

LangChain集成

OpenSearch MCP服务器可以使用SSE服务器传输轻松地与LangChain集成

先决条件

  1. 安装所需的软件包
pip install langchain langchain-mcp-adapters langchain-openai
  1. 设置OpenAI API密钥
export OPENAI_API_KEY="<your-openai-key>"
  1. 确保OpenSearch MCP服务器在SSE模式下运行
python -m mcp_server_opensearch --transport sse

集成脚本示例

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langchain.agents import AgentType, initialize_agent

# Initialize LLM (can use any LangChain-compatible LLM)
model = ChatOpenAI(model="gpt-4o")

async def main():
    # Connect to MCP server and create agent
    async with MultiServerMCPClient({
        "opensearch-mcp-server": {
            "transport": "sse",
            "url": "http://localhost:9900/sse",  # SSE server endpoint
            "headers": {
                "Authorization": "Bearer secret-token",
            }
        }
    }) as client:
        tools = client.get_tools()
        agent = initialize_agent(
            tools=tools,
            llm=model,
            agent=AgentType.OPENAI_FUNCTIONS,
            verbose=True,  # Enables detailed output of the agent's thought process
        )

        # Example query
        await agent.ainvoke({"input": "List all indices"})

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

笔记：

  • 该脚本与任何与LangChain集成并支持工具调用的LLM兼容
  • 在执行脚本之前，请确保OpenSearch MCP服务器正在运行
  • 根据需要配置身份验证和环境变量

发展

有兴趣贡献吗？查看我们的：

相关 MCP