自然言語でElasticsearchを操作！Python×Claude Desktop×MCP連携例

curl -LsSf https://astral.sh/uv/install.sh | sh

which uv

uv init mcp-elasticsearch

cd mcp-elasticsearch

uv add "mcp[cli]"

ELASTIC_USER=elastic
ELASTIC_PASSWORD=****
ELASTIC_CLOUD_URL=https://your-deployment.aws.found.io

# =========================================
# Elasticsearch MCP Server for Claude
# =========================================

from mcp.server.fastmcp import FastMCP
from elasticsearch import Elasticsearch
from dotenv import load_dotenv
import os

# ----------------------
# 🔧 Setup
# ----------------------

print("Loading environment variables...")
load_dotenv()

# Start MCP server
mcp = FastMCP("ElasticsearchMCP")

# Connect to Elasticsearch
try:
    es = Elasticsearch(
        os.getenv("ELASTIC_CLOUD_URL"),
        basic_auth=(os.getenv("ELASTIC_USER"), os.getenv("ELASTIC_PASSWORD"))
    )
    print("🔗 Connected to Elasticsearch!")
except Exception as e:
    print("❌ Failed to connect:", e)
    es = None

# ----------------------
# 🛠️ MCP Tools
# ----------------------

@mcp.tool()
def list_indices() -> list:
    """Get all index names in the cluster."""
    # ES Query: GET /_cat/indices
    return list(es.indices.get_alias().keys()) if es else ["ES not initialized"]

@mcp.tool()
def get_cluster_health() -> dict:
    """Check the health status of the Elasticsearch cluster."""
    # ES Query: GET /_cluster/health
    return es.cluster.health() if es else {"error": "ES not initialized"}

@mcp.tool()
def count_documents(index_name: str) -> int:
    """Count how many documents exist in a given index."""
    # ES Query: GET /{index}/_count
    try:
        return es.count(index=index_name)['count'] if es else -1
    except Exception as e:
        return f"Error: {str(e)}"

@mcp.tool()
def search_index(index_name: str, keyword: str) -> list:
    """Search for documents in an index using a keyword."""
    # ES Query:
    # POST /{index}/_search
    # {
    #   "query": { "match": { "_all": "keyword" } }
    # }
    try:
        result = es.search(
            index=index_name,
            query={"match": {"_all": keyword}},
            size=5
        )
        return [doc["_source"] for doc in result["hits"]["hits"]]
    except Exception as e:
        return [f"Error: {str(e)}"]

@mcp.tool()
def get_index_mapping(index_name: str) -> dict:
    """Retrieve the field definitions (mappings) of an index."""
    # ES Query: GET /{index}/_mapping
    try:
        return es.indices.get_mapping(index=index_name)[index_name]["mappings"]
    except Exception as e:
        return {"error": str(e)}

@mcp.tool()
def delete_index(index_name: str) -> str:
    """Delete an entire index (irreversible)."""
    # ES Query: DELETE /{index}
    try:
        es.indices.delete(index=index_name)
        return f"✅ Deleted index '{index_name}'"
    except Exception as e:
        return f"❌ Error: {str(e)}"

@mcp.tool()
def get_recent_docs(index_name: str, size: int = 5) -> list:
    """Fetch the most recent documents from an index."""
    # ES Query:
    # POST /{index}/_search
    # {
    #   "size": 5,
    #   "sort": [{ "@timestamp": "desc" }]
    # }
    try:
        result = es.search(
            index=index_name,
            size=size,
            sort=[{"@timestamp": "desc"}]
        )
        return [doc["_source"] for doc in result["hits"]["hits"]]
    except Exception as e:
        return [f"Error: {str(e)}"]

# ----------------------
# Start MCP Server
# ----------------------

if __name__ == "__main__":
    print("🚀 Starting MCP server...")
    if os.environ.get("RUN_MCP_MANUALLY"):
        print("Running in manual mode (not Claude Desktop).")
    else:
        print("Waiting for Claude Desktop input (stdio mode).")

    try:
        mcp.run()
        print("✅ MCP server exited cleanly.")
    except Exception as e:
        print("❌ MCP server crashed:", e)

uv run mcp install main.py

{
  "mcpServers": {
    "ElasticsearchMCP": {
      "command": "/Users/sam-koyama/.local/bin/uv",
      "args": [
        "--directory",
        "/Users/sam-koyama/MCP/second/mcp-elasticsearch",
        "run",
        "main.py"
      ],
      "env": {
        "ELASTIC_USER": "elastic",
        "ELASTIC_PASSWORD": "3fik3ua3kXnbOlp0ThX41be1",
        "ELASTIC_CLOUD_URL": "https://my-deployment-a17327.es.ap-northeast-1.aws.found.io"
      }
    }
  }
}