使用说明
OpenSearch MCP服务器-代码搜索代理
一本全面的指南，介绍如何设置OpenSearch进行代码搜索、获取源代码并将其与MCP（模型上下文协议）服务器集成，以实现AI驱动的代码搜索功能。
目录
概述
这个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=<your-strong-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:<your-password> https://localhost:9200
检查群集运行状况：
curl -k -u admin:<your-password> 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
}
设置索引
本指南为不同的用例使用了两个索引：
- 仅代码搜索文本：用于基于关键字的搜索
- 用向量进行代码搜索：用于混合搜索（关键字+语义）
索引定义文件
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:<your-password> \
-X PUT "https://localhost:9200/code-search-text-only" \
-H "Content-Type: application/json" \
-d @code-search-text-only.json
创建启用矢量的索引：
curl -k -u admin:<your-password> \
-X PUT https://localhost:9200/code-search-with-vectors \
-H "Content-Type: application/json" \
-d @code-search-with-vectors.json
验证索引创建
检查纯文本索引的映射：
curl -k -u admin:<your-password> https://localhost:9200/code-search-text-only/_mapping?pretty
检查启用矢量索引的映射：
curl -k -u admin:<your-password> 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 < len(lines):
start = i
end = min(i + LINE_CHUNK_SIZE, len(lines))
chunk = "\n".join(lines[start:end])
chunks.append({
"text": chunk,
"start_line": start + 1,
"end_line": end
})
i += (LINE_CHUNK_SIZE - LINE_CHUNK_OVERLAP)
return chunks
# -----------------------------
# Ingestion
# -----------------------------
def walk_repo(repo_path):
repo_name = os.path.basename(os.path.abspath(repo_path))
actions = []
for root, _, files in os.walk(repo_path):
for file in files:
full_path = os.path.join(root, file)
rel_path = os.path.relpath(full_path, repo_path)
if should_skip(full_path):
continue
language = detect_language(full_path)
if not language:
continue
try:
with open(full_path, "r", encoding="utf-8", errors="ignore") as f:
code = f.read()
except Exception:
continue
# 1️⃣ Symbol chunking
chunks = extract_symbols(code, language)
# 2️⃣ Fallback to line chunking
if not chunks:
chunks = line_chunks(code)
for chunk_id, chunk in enumerate(chunks):
actions.append({
"_index": INDEX_NAME,
"_source": {
"repo": repo_name,
"path": rel_path,
"language": language,
"chunk_id": chunk_id,
"start_line": chunk["start_line"],
"end_line": chunk["end_line"],
"content": chunk["text"]
}
})
if len(actions) >= 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("<your repo path>")
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 < len(lines):
start = i
end = min(i + LINE_CHUNK_SIZE, len(lines))
chunk = "\n".join(lines[start:end])
chunks.append({
"text": chunk,
"start_line": start + 1,
"end_line": end
})
i += (LINE_CHUNK_SIZE - LINE_CHUNK_OVERLAP)
return chunks
# -----------------------------
# Ingestion
# -----------------------------
def walk_repo(repo_path):
repo_name = os.path.basename(os.path.abspath(repo_path))
actions = []
for root, _, files in os.walk(repo_path):
for file in files:
full_path = os.path.join(root, file)
rel_path = os.path.relpath(full_path, repo_path)
if should_skip(full_path):
continue
language = detect_language(full_path)
if not language:
continue
try:
with open(full_path, "r", encoding="utf-8", errors="ignore") as f:
code = f.read()
except Exception:
continue
# 1️⃣ Symbol chunking
chunks = extract_symbols(code, language)
# 2️⃣ Fallback to line chunking
if not chunks:
chunks = line_chunks(code)
for chunk_id, chunk in enumerate(chunks):
# ---- Embedding generation ----
embedding = embedding_model.encode(
chunk["text"],
normalize_embeddings=True
).tolist()
actions.append({
"_index": INDEX_NAME,
"_source": {
"repo": repo_name,
"path": rel_path,
"language": language,
"chunk_id": chunk_id,
"start_line": chunk["start_line"],
"end_line": chunk["end_line"],
"content": chunk["text"],
"embedding": embedding
}
})
if len(actions) >= 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("<your repo path>")
跑步摄入
将代码摄入纯文本索引：
python ingest-code-text-only.py
在向量索引中嵌入代码：
python ingest-code-with-vectors.py
确认摄入
统计纯文本索引中的文档：
curl -k -u admin:<your-password> https://localhost:9200/code-search-text-only/_count
计数启用矢量索引的文档：
curl -k -u admin:<your-password> https://localhost:9200/code-search-with-vectors/_count
预期响应：
{
"count": 1234,
"_shards": {
"total": 2,
"successful": 2,
"skipped": 0,
"failed": 0
}
}
搜索代码
纯文本索引上的CLI搜索
搜索特定函数名称：
curl -k -u admin:<your-password> \
-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:<your-password> \
-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:<your-password> \
-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:<your-password> \
-X GET https://localhost:9200/code-search-with-vectors/_search \
-H "Content-Type: application/json" \
-d '{
"query": {
"bool": {
"should": [
{
"match": {
"content": "authentication handler"
}
}
]
}
}
}'
OpenSearch仪表板
访问仪表板
- 打开浏览器并导航到：
https://localhost:5601 - 使用以下方式登录：
- 用户名：
admin - 密码：
<your-password>
- 用户名：
仪表板功能
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桌面应用程序）以连接：
- 添加指向的服务器配置
http://localhost:8003 - 使用适当的凭据进行身份验证
- 使用自然语言搜索代码
对话示例：
User: "Find functions that implement retry logic"
Claude: [Uses search_code tool] "I found 3 code snippets that implement retry mechanisms..."
搜索策略
MCP服务器使用 混合搜索:
- 语义搜索：使用嵌入来理解查询意图
- 关键词搜索：使用2x boost执行精确的关键字匹配
- 排名：结合两种方法的分数
- 过滤：可选择按存储库筛选
维护
删除索引
要完全重置和删除索引，请执行以下操作：
curl -k -u admin:<your-password> -X DELETE https://localhost:9200/code-search-text-only
curl -k -u admin:<your-password> -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:<your-password> \
-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:<your-password> https://localhost:9200
问题：身份验证失败
解决方案：验证凭据是否与中的凭据匹配 .env 文件和检查日志：
docker compose logs opensearch
问题：搜索中没有结果
解决方案：验证数据是否被摄入：
curl -k -u admin:<your-password> https://localhost:9200/code-search-text-only/_count
问题：MCP服务器连接错误
解决方案：检查服务器是否在端口8003上运行：
curl http://localhost:8003
后续步骤
- 自定义摄入：修改摄取脚本以匹配您的代码库结构
- 调整搜索参数：调整块大小并提高因子以获得更好的结果
- 监测性能：使用OpenSearch Dashboard跟踪查询性能
- 扩大规模：为更大的代码库配置多个分片和副本
- 高级分析：构建自定义仪表板以分析代码指标
参考文献
许可证
该项目按原样提供，用于代码搜索和AI集成。
