Token导航 LogoToken导航TokenDH.com
中转站排行榜
Open Search MCP Server logo
开发工具stdio官方级别未说明来源级核验

Open Search MCP Server

MCP Server

OpenSearch MCP Server 是一个基于OpenSearch的代码搜索服务，提供语义和关键字混合搜索能力，适用于AI代理和开发人员快速查找代码片段。

工具数

2

提示词数

0

GitHub Stars

0

资源数

0
搜索代码搜索PythonClaude开发工具Claude

安装说明

本站只整理中文说明和来源信息，不托管安装包，也不代用户安装。

GitHub

作者 / 组织

yogeshsonawane-dev

提供方

yogeshsonawane-dev

最后核验

2026/5/17 20:21

快速接入

先看主来源和安装命令，再打开仓库或文档；下面只保留这个条目的关键接入事实。

命令预览

pip install -r requirements.txt

详细介绍

OpenSearch MCP服务器-代码搜索代理

一本全面的指南，介绍如何设置OpenSearch进行代码搜索、获取源代码并将其与MCP（模型上下文协议）服务器集成，以实现AI驱动的代码搜索功能。

目录

  1. 概述
  2. 先决条件
  3. 安装OpenSearch
  4. 设置索引
  5. 摄入代码
  6. 搜索代码
  7. OpenSearch仪表板
  8. MCP服务器集成
  9. 建筑

______________________________________________________________________

概述

这个OpenSearch MCP服务器使AI代理能够使用语义和基于关键字的搜索来搜索代码库。它提供：

  • 混合搜索：将语义理解（通过嵌入）与关键字匹配相结合
  • 双指数：用于快速关键字搜索的纯文本索引和用于语义搜索的矢量索引
  • 存储库筛选：需要时按存储库筛选结果
  • 代码元数据：检索文件路径、语言、行号和代码段
  • 与AI集成：通过MCP协议与Claude和其他AI模型无缝集成

______________________________________________________________________

先决条件

在开始之前，请确保您已经：

  • 码头工人：用于运行OpenSearch
  • Docker Compose：用于编排容器
  • Python 3.14+：用于运行摄取脚本和MCP服务器
  • pip或uv：Python包管理器

______________________________________________________________________

安装OpenSearch

步骤1：创建环境文件

创建 .env 在项目根目录中使用OpenSearch初始管理员密码的文件：

echo "OPENSEARCH_INITIAL_ADMIN_PASSWORD=" > .env

步骤2：使用Docker Compose启动OpenSearch

从OpenSearch官方网站下载docker镜像，使用docker Compose启动OpenSearch和OpenSearch Dashboards：

docker compose up -d

步骤3：验证OpenSearch是否正在运行

检查所有容器是否正在运行：

docker ps

第四步：健康检查

验证OpenSearch是否响应：

curl -k -u admin: https://localhost:9200

检查群集运行状况：

curl -k -u admin: https://localhost:9200/_cluster/health

预期响应：

{
  "cluster_name": "opensearch-cluster",
  "status": "green",
  "timed_out": false,
  "number_of_nodes": 1,
  "number_of_data_nodes": 1,
  "active_primary_shards": 0,
  "active_shards": 0,
  "relocating_shards": 0,
  "initializing_shards": 0,
  "unassigned_shards": 0,
  "delayed_unassigned_shards": 0,
  "number_of_pending_tasks": 0,
  "number_of_in_flight_fetch": 0,
  "task_max_waiting_in_queue_ms": 0,
  "active_shards_percent_as_number": 100.0
}

______________________________________________________________________

设置索引

本指南为不同的用例使用了两个索引：

  1. 仅代码搜索文本：用于基于关键字的搜索
  2. 用向量进行代码搜索：用于混合搜索（关键字+语义）

索引定义文件

code-search-text-only.json

{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 0,
    "analysis": {
      "tokenizer": {
        "code_tokenizer": {
          "type": "pattern",
          "pattern": "[^a-zA-Z0-9_]"
        }
      },
      "filter": {
        "limit_token_length": {
          "type": "length",
          "max": 256
        }
      },
      "analyzer": {
        "code_analyzer": {
          "type": "custom",
          "tokenizer": "code_tokenizer",
          "filter": [
            "lowercase",
            "limit_token_length"
          ]
        }
      }
    }
  },
  "mappings": {
    "dynamic": "false",
    "properties": {
      "repo": {
        "type": "keyword"
      },
      "path": {
        "type": "keyword"
      },
      "language": {
        "type": "keyword"
      },
      "chunk_id": {
        "type": "integer"
      },
      "content": {
        "type": "text",
        "analyzer": "code_analyzer"
      }
    }
  }
}

code-search-with-vectors.json

{
  "settings": {
    "index.knn": true,
    "number_of_shards": 1,
    "number_of_replicas": 0,
    "analysis": {
      "tokenizer": {
        "code_tokenizer": {
          "type": "pattern",
          "pattern": "[^a-zA-Z0-9_]"
        }
      },
      "filter": {
        "limit_token_length": {
          "type": "length",
          "max": 256
        }
      },
      "analyzer": {
        "code_analyzer": {
          "type": "custom",
          "tokenizer": "code_tokenizer",
          "filter": [
            "lowercase",
            "limit_token_length"
          ]
        }
      }
    }
  },
  "mappings": {
    "dynamic": "false",
    "properties": {
      "repo": { "type": "keyword" },
      "path": { "type": "keyword" },
      "language": { "type": "keyword" },
      "chunk_id": { "type": "integer" },
      "start_line": { "type": "integer" },
      "end_line": { "type": "integer" },

      "content": {
        "type": "text",
        "analyzer": "code_analyzer"
      },

      "embedding": {
        "type": "knn_vector",
        "dimension": 384,
        "method": {
          "name": "hnsw",
          "engine": "nmslib",
          "space_type": "cosinesimil"
        }
      }
    }
  }
}

创建索引

创建纯文本索引：

curl -k -u admin: \
  -X PUT "https://localhost:9200/code-search-text-only" \
  -H "Content-Type: application/json" \
  -d @code-search-text-only.json

创建启用矢量的索引：

curl -k -u admin: \
  -X PUT https://localhost:9200/code-search-with-vectors \
  -H "Content-Type: application/json" \
  -d @code-search-with-vectors.json

验证索引创建

检查纯文本索引的映射：

curl -k -u admin: https://localhost:9200/code-search-text-only/_mapping?pretty

检查启用矢量索引的映射：

curl -k -u admin: https://localhost:9200/code-search-with-vectors/_mapping?pretty

______________________________________________________________________

摄入代码

Python依赖关系

安装所需的依赖项：

pip install -r requirements.txt

或使用紫外线：

uv pip install -r requirements.txt

摄入脚本

您需要两个摄取脚本：

ingest-code-ext-only.py

此脚本将代码摄取到纯文本索引中：

import os
from dotenv import load_dotenv
from opensearchpy import OpenSearch, helpers
from tree_sitter_languages import get_parser

# Load environment variables from .env file
load_dotenv()

# -----------------------------
# OpenSearch configuration
# -----------------------------
opensearch_password = os.getenv("OPENSEARCH_INITIAL_ADMIN_PASSWORD")
if not opensearch_password:
    raise ValueError("OPENSEARCH_INITIAL_ADMIN_PASSWORD not found in .env file")

client = OpenSearch(
    hosts=[{"host": "localhost", "port": 9200}],
    http_auth=("admin", opensearch_password),
    use_ssl=True,
    verify_certs=False,
    ssl_show_warn=False,
)

INDEX_NAME = "code-search-text-only"

# -----------------------------
# Config
# -----------------------------
LANGUAGE_MAP = {
    ".java": "java",
    ".py": "python",
    ".js": "javascript",
    ".ts": "typescript",
    ".go": "go",
    ".rb": "ruby",
    ".rs": "rust",
    ".c": "c",
    ".h": "c",
    ".cpp": "cpp",
    ".hpp": "cpp",
    ".cs": "c_sharp",
    '.php': 'php',
    '.swift': 'swift',
    '.kt': 'kotlin',
    '.scala': 'scala',
}

SKIP_EXTENSIONS = (
    ".min.js", ".map", ".lock", ".zip", ".jar", ".class",
    ".png", ".jpg", ".jpeg", ".gif", ".pdf"
)

MAX_FILE_SIZE = 500_000        # 500 KB
LINE_CHUNK_SIZE = 200          # lines
LINE_CHUNK_OVERLAP = 40

# -----------------------------
# Helpers
# -----------------------------
def detect_language(path):
    for ext, lang in LANGUAGE_MAP.items():
        if path.endswith(ext):
            return lang
    return None

def should_skip(path):
    if path.lower().endswith(SKIP_EXTENSIONS):
        return True
    if os.path.getsize(path) > MAX_FILE_SIZE:
        return True
    return False

# -----------------------------
# Tree-sitter symbol extraction
# -----------------------------
SYMBOL_NODES = {
    "function_definition",
    "method_definition",
    "method_declaration",
    "function_declaration",
    "class_definition",
    "class_declaration"
}

def extract_symbols(code, language):
    parser = get_parser(language)
    tree = parser.parse(code.encode("utf8"))
    root = tree.root_node
    symbols = []

    def walk(node):
        if node.type in SYMBOL_NODES:
            start = node.start_byte
            end = node.end_byte
            symbols.append({
                "text": code[start:end],
                "start_line": node.start_point[0] + 1,
                "end_line": node.end_point[0] + 1
            })
        for child in node.children:
            walk(child)

    walk(root)
    return symbols

# -----------------------------
# Line fallback chunking
# -----------------------------
def line_chunks(code):
    lines = code.splitlines()
    chunks = []

    i = 0
    while i = 500:
                helpers.bulk(client, actions)
                actions.clear()

    if actions:
        helpers.bulk(client, actions)

    print(f"Indexing complete for repo: {repo_name}")

# -----------------------------
# Run
# -----------------------------
if __name__ == "__main__":
    walk_repo("")

ingest-code with sectors.py

此脚本将代码嵌入到启用向量的索引中：

import os
from dotenv import load_dotenv
from opensearchpy import OpenSearch, helpers
from tree_sitter_languages import get_parser
from sentence_transformers import SentenceTransformer

# Load environment variables from .env file
load_dotenv()

# -----------------------------
# OpenSearch configuration
# -----------------------------
opensearch_password = os.getenv("OPENSEARCH_INITIAL_ADMIN_PASSWORD")
if not opensearch_password:
    raise ValueError("OPENSEARCH_INITIAL_ADMIN_PASSWORD not found in .env file")

client = OpenSearch(
    hosts=[{"host": "localhost", "port": 9200}],
    http_auth=("admin", opensearch_password),
    use_ssl=True,
    verify_certs=False,
    ssl_show_warn=False,
)

INDEX_NAME = "code-search-with-vectors"

# -----------------------------
# Embedding model (loaded once)
# -----------------------------
EMBEDDING_MODEL_NAME = "all-MiniLM-L6-v2"
embedding_model = SentenceTransformer(EMBEDDING_MODEL_NAME)

# -----------------------------
# Config
# -----------------------------
LANGUAGE_MAP = {
    ".java": "java",
    ".py": "python",
    ".js": "javascript",
    ".ts": "typescript",
    ".go": "go",
    ".rb": "ruby",
    ".rs": "rust",
    ".c": "c",
    ".h": "c",
    ".cpp": "cpp",
    ".hpp": "cpp",
    ".cs": "c_sharp",
    '.php': 'php',
    '.swift': 'swift',
    '.kt': 'kotlin',
    '.scala': 'scala',
}

SKIP_EXTENSIONS = (
    ".min.js", ".map", ".lock", ".zip", ".jar", ".class",
    ".png", ".jpg", ".jpeg", ".gif", ".pdf"
)

MAX_FILE_SIZE = 500_000        # 500 KB
LINE_CHUNK_SIZE = 200          # lines
LINE_CHUNK_OVERLAP = 40

# -----------------------------
# Helpers
# -----------------------------
def detect_language(path):
    for ext, lang in LANGUAGE_MAP.items():
        if path.endswith(ext):
            return lang
    return None

def should_skip(path):
    if path.lower().endswith(SKIP_EXTENSIONS):
        return True
    if os.path.getsize(path) > MAX_FILE_SIZE:
        return True
    return False

# -----------------------------
# Tree-sitter symbol extraction
# -----------------------------
SYMBOL_NODES = {
    "function_definition",
    "method_definition",
    "method_declaration",
    "function_declaration",
    "class_definition",
    "class_declaration"
}

def extract_symbols(code, language):
    parser = get_parser(language)
    tree = parser.parse(code.encode("utf8"))
    root = tree.root_node
    symbols = []

    def walk(node):
        if node.type in SYMBOL_NODES:
            start = node.start_byte
            end = node.end_byte
            symbols.append({
                "text": code[start:end],
                "start_line": node.start_point[0] + 1,
                "end_line": node.end_point[0] + 1
            })
        for child in node.children:
            walk(child)

    walk(root)
    return symbols

# -----------------------------
# Line fallback chunking
# -----------------------------
def line_chunks(code):
    lines = code.splitlines()
    chunks = []

    i = 0
    while i = 500:
                helpers.bulk(client, actions)
                actions.clear()

    if actions:
        helpers.bulk(client, actions)

    print(f"Indexing complete for repo: {repo_name}")

# -----------------------------
# Run
# -----------------------------
if __name__ == "__main__":
    walk_repo("")

跑步摄入

将代码摄入纯文本索引：

python ingest-code-text-only.py

在向量索引中嵌入代码：

python ingest-code-with-vectors.py

确认摄入

统计纯文本索引中的文档：

curl -k -u admin: https://localhost:9200/code-search-text-only/_count

计数启用矢量索引的文档：

curl -k -u admin: https://localhost:9200/code-search-with-vectors/_count

预期响应：

{
  "count": 1234,
  "_shards": {
    "total": 2,
    "successful": 2,
    "skipped": 0,
    "failed": 0
  }
}

______________________________________________________________________

搜索代码

纯文本索引上的CLI搜索

搜索特定函数名称：

curl -k -u admin: \
  -X GET https://localhost:9200/code-search-text-only/_search \
  -H "Content-Type: application/json" \
  -d '{
    "query": {
      "match": {
        "content": "getGTTs"
      }
    }
  }'

搜索API终结点：

curl -k -u admin: \
  -X GET https://localhost:9200/code-search-text-only/_search \
  -H "Content-Type: application/json" \
  -d '{
    "query": {
      "match": {
        "content": "/shops/{shopId}"
      }
    }
  }'

矢量索引的CLI搜索

搜索语义匹配：

curl -k -u admin: \
  -X GET https://localhost:9200/code-search-with-vectors/_search \
  -H "Content-Type: application/json" \
  -d '{
    "query": {
      "match": {
        "content": "retry mechanism"
      }
    }
  }'

高级查询：混合搜索

结合关键字和语义搜索（例如 code-search-with-vectors):

curl -k -u admin: \
  -X GET https://localhost:9200/code-search-with-vectors/_search \
  -H "Content-Type: application/json" \
  -d '{
    "query": {
      "bool": {
        "should": [
          {
            "match": {
              "content": "authentication handler"
            }
          }
        ]
      }
    }
  }'

______________________________________________________________________

OpenSearch仪表板

访问仪表板

  1. 打开浏览器并导航到： https://localhost:5601
  2. 使用以下方式登录：

- 用户名： admin - 密码： ``

仪表板功能

1.指标管理

  • 导航到 堆栈管理 → 索引管理
  • 查看所有索引及其运行状况
  • 监控分片分配和文档计数

2.开发工具

  • 首选 管理 → 开发者工具
  • 使用控制台运行Elasticsearch/OpenSearch查询
  • 以交互方式测试您的搜索查询

3.在仪表板中搜索查询

Console中的搜索示例：

GET /code-search-text-only/_search
{
  "query": {
    "match": {
      "content": "database"
    }
  },
  "size": 20
}

4.可视化

  • 从搜索结果创建可视化
  • 构建仪表板以监控代码搜索指标
  • 跟踪摄取进度和文档计数

______________________________________________________________________

MCP服务器集成

什么是MCP服务器？

这 模型上下文协议（MCP）服务器 是一个集成层，将您的OpenSearch实例与Claude等AI模型连接起来。它能够：

  • 自然语言代码搜索查询
  • 自动将查询转换为OpenSearch查询
  • 结果格式和排名
  • 与AI驱动的代码分析工具集成

建筑

┌─────────────┐
│   Claude/   │
│     AI      │
└──────┬──────┘
       │ (MCP Protocol)
       ▼
┌──────────────────────┐
│  MCP Server (main.py)│
│ - search_code()      │
│ - code_search_agent()│
└──────┬───────────────┘
       │ (HTTP/REST)
       ▼
┌──────────────────────┐
│   OpenSearch         │
│  - Vector Search     │
│  - Keyword Search    │
└──────────────────────┘

安装

安装Python依赖项：

pip install -r requirements.txt

或使用紫外线：

uv sync

配置

MCP服务器使用以下方式连接到OpenSearch：

  • 主机: localhost
  • 端口: 9200
  • 认证：管理员凭据（在main.py中配置）
  • 索引: code-search-with-vectors （混合搜索索引）

运行MCP服务器

启动MCP服务器：

python main.py

预期产量：

Starting MCP Server on http://localhost:8003

服务器功能

MCP服务器提供：

1.搜索编码工具

参数：

  • query （字符串，必填）：自然语言搜索查询
  • repo （字符串，可选）：按存储库筛选结果

退货：

  • 包含元数据的代码段列表：

- score：相关性得分（0-1） - repo：存储库名称 - path：文件路径 - language：编程语言 - start_line：起始行号 - end_line：结束行号 - content：代码片段

示例查询：

Find all functions that handle authentication

预期响应：

[
  {
    "score": 0.95,
    "repo": "my-repo",
    "path": "src/auth/handler.py",
    "language": "python",
    "start_line": 42,
    "end_line": 68,
    "content": "def handle_authentication(...)"
  }
]

2.code_search_agent提示

为AI代理提供系统说明，说明如何：

  • 有效地使用search_code工具
  • 用适当的上下文格式化响应
  • 切勿伪造代码片段
  • 处理边缘案例（未找到结果）

与Claude整合

MCP服务器运行后，配置您的MCP客户端（例如Claude桌面应用程序）以连接：

  1. 添加指向的服务器配置 http://localhost:8003
  2. 使用适当的凭据进行身份验证
  3. 使用自然语言搜索代码

对话示例：

User: "Find functions that implement retry logic"
Claude: [Uses search_code tool] "I found 3 code snippets that implement retry mechanisms..."

搜索策略

MCP服务器使用 混合搜索:

  1. 语义搜索：使用嵌入来理解查询意图
  2. 关键词搜索：使用2x boost执行精确的关键字匹配
  3. 排名：结合两种方法的分数
  4. 过滤：可选择按存储库筛选

______________________________________________________________________

维护

删除索引

要完全重置和删除索引，请执行以下操作：

curl -k -u admin: -X DELETE https://localhost:9200/code-search-text-only
curl -k -u admin: -X DELETE https://localhost:9200/code-search-with-vectors

停止OpenSearch

停止并移除容器：

docker compose down

删除卷（清除所有数据）：

docker compose down -v

备份数据

创建OpenSearch数据的备份：

docker compose exec opensearch curl -k -u admin: \
  -X POST "https://localhost:9200/_snapshot/backup" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "fs",
    "settings": {
      "location": "/backups"
    }
  }'

______________________________________________________________________

故障排除

问题：连接被拒绝

解决方案：确保OpenSearch运行正常：

docker ps
curl -k -u admin: https://localhost:9200

问题：身份验证失败

解决方案：验证凭据是否与中的凭据匹配 .env 文件和检查日志：

docker compose logs opensearch

问题：搜索中没有结果

解决方案：验证数据是否被摄入：

curl -k -u admin: https://localhost:9200/code-search-text-only/_count

问题：MCP服务器连接错误

解决方案：检查服务器是否在端口8003上运行：

curl http://localhost:8003

______________________________________________________________________

后续步骤

  1. 自定义摄入：修改摄取脚本以匹配您的代码库结构
  2. 调整搜索参数：调整块大小并提高因子以获得更好的结果
  3. 监测性能：使用OpenSearch Dashboard跟踪查询性能
  4. 扩大规模：为更大的代码库配置多个分片和副本
  5. 高级分析：构建自定义仪表板以分析代码指标

______________________________________________________________________

参考文献

______________________________________________________________________

许可证

该项目按原样提供，用于代码搜索和AI集成。

目录标签

目录标签

搜索代码搜索PythonClaude开发工具本地部署语义搜索AI集成OpenSearch

支持客户端

Claude

接入字段

传输方式（transport，传输协议）

stdio

鉴权方式（authType，认证方式）

none

工具数量（toolCount，工具数）

2

资源数量（resourceCount，资源数）

0

提示词数量（promptCount，提示词数）

0

权限和风险

stdionone部署方式未说明

接入前请确认传输方式、认证方式和部署位置，并根据实际工具能力限制访问范围。

安装前确认

不要直接授予不必要的文件、网络或账号权限；先核对安装命令和配置内容。

来源信息

open-search-mcp-server logo

MCP World

mcpworld.com

/detail/d9efe20c6e8bb3b65ed6933885319094

OpenSearch MCP Server 是一个基于OpenSearch的代码搜索服务，提供语义和关键字混合搜索能力，适用于AI代理和开发人员快速查找代码片段。

继续浏览同类 MCP

更多开发工具 MCP（20,396）返回 MCP 目录

上一个 MCP

Openscad MCP Server 9980e73a

Provides OpenSCAD rendering capabilities through a Model Context Protocol interface, allowing users to generate PNG images from OpenSCAD code without a GUI.

下一个 MCP

OpenSincera MCP Server

Provides access to OpenSincera API for retrieving digital advertising publisher metadata, verification status, and operational metrics through domain or Publisher ID lookup.