customized-elasticsearch-mcp-server

Coolgiserz/customized-elasticsearch-mcp-server

3.3

If you are the rightful owner of customized-elasticsearch-mcp-server and would like to certify it and/or have it hosted online, please leave a comment on the right or send an email to henry@mcphub.com.

Customized-Elasticsearch-MCP-Server is a Python-based server leveraging FastMCP and FastAPI to provide a tailored Elasticsearch MCP service.

The Customized-Elasticsearch-MCP-Server is a specialized server application designed to facilitate efficient interaction with Elasticsearch through the MCP (Model Context Protocol) framework. Built using Python's FastAPI and FastMCP, this server offers a robust solution for handling search queries, secondary filtering, and ID-based lookups within Elasticsearch indices. It integrates seamlessly with Prometheus for monitoring and uses Redis for session management, ensuring both performance and scalability. The server is containerized using Docker, making it easy to deploy and manage across different environments. With a focus on modularity and extensibility, the server supports various MCP tools for specific tasks, and its architecture is designed to accommodate future enhancements such as advanced authentication mechanisms and query optimizations.

Features

  • Keyword search, secondary filtering, and ID-based queries
  • Integration with Prometheus for monitoring
  • Session management using Redis
  • Docker and Docker Compose support for easy deployment
  • Comprehensive testing suite including unit and integration tests

Usages

local development

python
mcp.run(transport='stdio')  # Tools defined via @mcp.tool() decorator

remote deployment

python
mcp.run(transport='sse', host="0.0.0.0", port=8000)  # Specify SSE endpoint

docker compose

bash
make docker-build
make docker-up

docker

bash
make build

fastmcp client usage

python
import asyncio
from fastmcp import Client

async def main():
    # 将 URL 替换为实际部署地址
    async with Client("http://localhost:9009/mcp-server") as client:
        tools = await client.list_tools()
        print("可用工具:", tools)

        result = await client.call_tool("search_news", {
            "query": "人工智能",
            "max_results": 5,
            "date_from": "2024-01-01",
            "date_to": "2024-12-31"
        })
        print(result)

asyncio.run(main())

Tools

  1. search_news

    Tool for searching news articles based on keywords.

  2. search_news_with_secondary_filter

    Tool for searching news with additional filtering options.

  3. read_single_news

    Tool for retrieving a single news article by ID.