OpenSearch MCP Server

お知らせ：このプロジェクトは成熟段階に達し、opensearch-mcp-server-py リポジトリに移行しました。そちらでお会いしましょう！このリポジトリは現在アーカイブされています。

OpenSearch MCP Server

stdio および sse サーバー経由で 4 つのツールを公開する、OpenSearch 用の最小限の Model Context Protocol（MCP）サーバーです。

利用可能なツール

• 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 Desktop との連携

• 公開されている 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

2. OpenAI API キーを設定します

export OPENAI_API_KEY="<your-openai-key>"

3. 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 サーバーが実行されていることを確認してください。
• 必要に応じて、認証と環境変数を設定してください。

開発

貢献に興味がありますか？以下をご覧ください：

• 開発ガイド - 開発環境をセットアップします
• 貢献ガイドライン - 貢献方法を学びます