使用 TypeScript 创建 Elasticsearch MCP 服务器：让 Claude Desktop 直连检索

1. 从零搭建 TypeScript Elasticsearch MCP 服务器为什么值得自己写一个如果你手里有一批技术文档、内部规范或者产品手册平时用 Claude Desktop 聊天时总希望它能直接翻你的 Elasticsearch 索引来回答问题而不是靠模型自己编那 MCPModel Context Protocol就是当前最顺手的路子。MCP 是 Anthropic 提出的开放标准专门用来让大语言模型和外部系统之间建立安全的双向连接。你可以把它理解成给 Claude 装了一个插件接口只要按协议暴露工具Claude 就能在对话里自动调用。Elastic 官方其实已经提供了 Agent Builder 的 MCP endpoint也有 Python 版本的 MCP 服务器。但官方方案有个明显限制Agent Builder 的 endpoint 基本只走 ES|QL 查询你想用完整的 Query DSL、想自己控制结果怎么格式化、想在返回给模型之前再插一步摘要或过滤就不太自由了。自己用 TypeScript 写一个 MCP 服务器好处就在这——搜索逻辑、字段权重、fuzziness、结果裁剪、引用格式全都你说了算。这篇要做的是一个能跑通的完整链路用 TypeScript 写一个 MCP 服务器暴露两个工具一个负责在 Elasticsearch 里做全文检索另一个负责把检索结果交给模型做摘要并附上引用来源最后在 Claude Desktop 里完成接入和端到端验证。适合谁适合已经有一份 Elasticsearch 数据、想让 AI 客户端直接查这份数据的后端或全栈工程师。前置条件不复杂Node.js 20 以上、一个能访问的 Elasticsearch 实例、一个 OpenAI API Key用于摘要那一步、以及装好的 Claude Desktop。整个项目结构很轻核心就是一个index.ts编译后产出dist/index.jsClaude Desktop 通过 stdio 把它作为子进程拉起来。下面按初始化项目 → 写服务器 → 定义工具 → 编译 → 接入 Claude Desktop → 验证的顺序走一遍每一步都给可复制的命令和代码。2. 初始化项目与依赖TypeScript Elasticsearch MCP 服务器环境搭建先把工程骨架搭起来。新建一个目录进去之后初始化 Node 应用mkdir es-mcp-server cd es-mcp-server npm init -y这一步会生成package.json。接着装运行依赖和开发依赖。运行依赖有四个elastic/elasticsearch负责和 Elasticsearch 通信modelcontextprotocol/sdk提供创建 MCP 服务器、注册工具、和客户端通信的核心能力openai用来调模型做摘要zod用来给每个工具的输入输出定义结构化 schema 并在运行时校验。npm install elastic/elasticsearch modelcontextprotocol/sdk openai zod npm install --save-dev ts-node types/node typescript装完之后建议在package.json里补一个type字段和编译脚本避免后面模块解析出问题。把package.json改成类似这样{ name: es-mcp-server, version: 1.0.0, type: module, scripts: { build: tsc index.ts --target ES2022 --module node16 --moduleResolution node16 --outDir ./dist --strict --esModuleInterop, start: node ./dist/index.js }, dependencies: { elastic/elasticsearch: ^8.15.0, modelcontextprotocol/sdk: ^1.0.0, openai: ^4.60.0, zod: ^3.23.8 }, devDependencies: { types/node: ^20.14.0, ts-node: ^10.9.2, typescript: ^5.5.0 } }这里type: module很关键。MCP 的 SDK 用的是 ESM 风格的导入路径比如modelcontextprotocol/sdk/server/mcp.js如果项目还是 CommonJS导入时会报模块找不到。module和moduleResolution都设成node16配合ES2022目标能正确处理.js后缀的 ESM 导入。关于数据集为了演示方便我们假设索引名叫documents每条文档长这样{ id: 5, title: Logging Standards for Microservices, content: Consistent logging across microservices helps with debugging and tracing. Use structured JSON logs and include request IDs and timestamps. Avoid logging sensitive information. Centralize logs in Elasticsearch or a similar system., tags: [logging, microservices, standards] }你可以自己写一个简单的摄取脚本用elastic/elasticsearch的client.index()把一批这样的文档灌进去或者用_bulk批量导入。索引的 mapping 里title和content用text类型tags用keyword这样后面的multi_match才能正常工作。环境变量方面我们约定三个ELASTICSEARCH_ENDPOINT、ELASTICSEARCH_API_KEY、OPENAI_API_KEY代码里会从process.env读取Claude Desktop 的配置里再注入。3. 编写 MCP 服务器与工具定义可复制的 index.ts 配置现在写核心文件index.ts。先导入依赖并处理环境变量和客户端初始化import { z } from zod; import { Client } from elastic/elasticsearch; import { McpServer } from modelcontextprotocol/sdk/server/mcp.js; import { StdioServerTransport } from modelcontextprotocol/sdk/server/stdio.js; import OpenAI from openai; const ELASTICSEARCH_ENDPOINT process.env.ELASTICSEARCH_ENDPOINT ?? http://localhost:9200; const ELASTICSEARCH_API_KEY process.env.ELASTICSEARCH_API_KEY ?? ; const OPENAI_API_KEY process.env.OPENAI_API_KEY ?? ; const INDEX documents; const openai new OpenAI({ apiKey: OPENAI_API_KEY }); const esClient new Client({ node: ELASTICSEARCH_ENDPOINT, auth: { apiKey: ELASTICSEARCH_API_KEY }, });用 zod 定义文档和搜索结果的 schema这样工具输入输出都能在运行时校验const DocumentSchema z.object({ id: z.number(), title: z.string(), content: z.string(), tags: z.array(z.string()), }); const SearchResultSchema z.object({ id: z.number(), title: z.string(), content: z.string(), tags: z.array(z.string()), score: z.number(), }); type Document z.infertypeof DocumentSchema; type SearchResult z.infertypeof SearchResultSchema;初始化 MCP 服务器const server new McpServer({ name: Elasticsearch RAG MCP, description: A RAG server using Elasticsearch. Provides tools for document search, result summarization, and source citation., version: 1.0.0, });第一个工具search_docs做全文检索。注意multi_match里title^2给标题加权fuzziness: AUTO提供拼写容错should里再加一个match_phrase提升短语匹配server.registerTool( search_docs, { title: Search Documents, description: Search for documents in Elasticsearch using full-text search. Returns the most relevant documents with their content, title, tags, and relevance score., inputSchema: { query: z.string().describe(The search query terms to find relevant documents), max_results: z.number().optional().default(5).describe(Maximum number of results to return), }, outputSchema: { results: z.array(SearchResultSchema), total: z.number(), }, }, async ({ query, max_results }) { if (!query) { return { content: [{ type: text, text: Query parameter is required }], isError: true }; } try { const response await esClient.search({ index: INDEX, size: max_results, query: { bool: { must: [ { multi_match: { query, fields: [title^2, content, tags], fuzziness: AUTO } }, ], should: [ { match_phrase: { title: { query, boost: 2 } } }, ], }, }, highlight: { fields: { title: {}, content: {} } }, }); const results: SearchResult[] response.hits.hits.map((hit: any) { const source hit._source as Document; return { id: source.id, title: source.title, content: source.content, tags: source.tags, score: hit._score ?? 0 }; }); const contentText results .map((r, i) [${i 1}] ${r.title} (score: ${r.score.toFixed(2)})\n${r.content.substring(0, 200)}...) .join(\n\n); const totalHits typeof response.hits.total number ? response.hits.total : (response.hits.total?.value ?? 0); return { content: [{ type: text, text: Found ${results.length} relevant documents:\n\n${contentText} }], structuredContent: { results, total: totalHits }, }; } catch (error: any) { return { content: [{ type: text, text: Error searching documents: ${error.message} }], isError: true }; } } );第二个工具summarize_and_cite把上一步的结果交给gpt-4o-mini做摘要同时返回引用元数据server.registerTool( summarize_and_cite, { title: Summarize and Cite, description: Summarize the provided search results to answer a question and return citation metadata for the sources used., inputSchema: { results: z.array(SearchResultSchema).describe(Array of search results from search_docs), question: z.string().describe(The question to answer), max_length: z.number().optional().default(500).describe(Maximum length of the summary in characters), max_docs: z.number().optional().default(5).describe(Maximum number of documents to include in the context), }, outputSchema: { summary: z.string(), sources_used: z.number(), citations: z.array(z.object({ id: z.number(), title: z.string(), tags: z.array(z.string()), relevance_score: z.number(), })), }, }, async ({ results, question, max_length, max_docs }) { if (!results || results.length 0 || !question) { return { content: [{ type: text, text: Both results and question parameters are required, and results must not be empty }], isError: true }; } try { const used results.slice(0, max_docs); const context used .map((r: SearchResult, i: number) [Document ${i 1}: ${r.title}]\n${r.content}) .join(\n\n---\n\n); const completion await openai.chat.completions.create({ model: gpt-4o-mini, messages: [ { role: system, content: You are a helpful assistant that answers questions based on provided documents. Synthesize information from the documents to answer the users question accurately and concisely. If the documents dont contain relevant information, say so. }, { role: user, content: Question: ${question}\n\nRelevant Documents:\n${context} }, ], max_tokens: Math.min(Math.ceil(max_length / 4), 1000), temperature: 0.3, }); const summaryText completion.choices[0]?.message?.content ?? No summary generated.; const citations used.map((r: SearchResult) ({ id: r.id, title: r.title, tags: r.tags, relevance_score: r.score, })); const citationText citations .map((c, i) [${i 1}] ID: ${c.id}, Title: ${c.title}, Tags: ${c.tags.join(, )}, Score: ${c.relevance_score.toFixed(2)}) .join(\n); return { content: [{ type: text, text: Summary:\n\n${summaryText}\n\nSources used (${citations.length}):\n\n${citationText} }], structuredContent: { summary: summaryText, sources_used: citations.length, citations }, }; } catch (error: any) { return { content: [{ type: text, text: Error generating summary and citations: ${error.message} }], isError: true }; } } );最后用 stdio 传输启动服务器。stdio 是最简单的传输方式客户端把服务器当子进程拉起通过标准输入输出通信const transport new StdioServerTransport(); server.connect(transport);编译npx tsc index.ts --target ES2022 --module node16 --moduleResolution node16 --outDir ./dist --strict --esModuleInterop编译成功后dist/index.js就是 Claude Desktop 要加载的入口文件。4. 接入 Claude Desktop 并验证一次索引查询的端到端测试打开 Claude Desktop 的配置文件macOS 一般在~/Library/Application Support/Claude/claude_desktop_config.jsonWindows 在%APPDATA%\Claude\claude_desktop_config.json加入我们的服务器{ mcpServers: { elasticsearch-rag-mcp: { command: node, args: [/Users/user-name/app-dir/dist/index.js], env: { ELASTICSEARCH_ENDPOINT: your-endpoint-here, ELASTICSEARCH_API_KEY: your-api-key-here, OPENAI_API_KEY: your-openai-key-here } } } }args指向编译后的dist/index.js绝对路径env里的变量名必须和代码里process.env读取的完全一致。改完保存重启 Claude Desktop。重启后在输入框附近点开 Search and Tools确认search_docs和summarize_and_cite两个工具都处于启用状态。如果弹出子菜单问你是否批准使用每个工具选 Always allow 或 Allow once。现在做一次端到端验证。在 Claude Desktop 里输入Search for documents about authentication methods and role-based access control.Claude 会先调用search_docs返回类似这样的结果Found 5 relevant documents: [1] Access Control and Role Management (score: 8.42) This document covers role-based access control (RBAC) principles, including ensuring users only have necessary permissions... [2] User Authentication with OAuth 2.0 (score: 7.15) This document explains OAuth 2.0 authentication, which enables secure delegated access without credential sharing...再试一个会触发链式调用的查询What are the main recommendations to improve authentication and access control across our systems? Include references.这时 Claude 会先调search_docs拿到文档再把结果传给summarize_and_cite最终返回带引用的摘要类似Based on the documentation, here are the main recommendations: 1. Implement Role-Based Access Control (RBAC) - Ensure users have only the permissions necessary for their job functions. [1] 2. Regular Access Audits - Conduct regular audits of user roles and promptly revoke access for inactive accounts. [1] 3. OAuth 2.0 for Secure Authentication - Use OAuth 2.0 to enable secure delegated access without sharing user credentials. [2] References [1] Access Control and Role Management (Tags: security, access-control) [2] User Authentication with OAuth 2.0 (Tags: authentication, oauth)看到这个带引用的结果说明整条链路——Claude Desktop → MCP 服务器 → Elasticsearch 检索 → OpenAI 摘要 → 引用回传——已经跑通。整个过程不需要你手动在中间传数据Claude 自己判断该调哪个工具、按什么顺序调。5. 常见报错排查401、local proxy failed、reading choices 怎么处理接入过程中最容易卡在几个地方这里按真实报错对照排查。401 Unauthorized 或 security_exception这是 Elasticsearch 认证失败。先确认ELASTICSEARCH_API_KEY是不是完整的 API Key有些控制台只显示一次复制不全就会 401。再确认 endpoint 带不带协议头http://localhost:9200和localhost:9200在elastic/elasticsearch里行为不同后者可能被当成非法 URL。如果你用的是 Elastic Cloudendpoint 应该是https://xxx.es.region.aws.elastic-cloud.com这种形式。另外检查 API Key 对应的角色有没有目标索引的read权限。local proxy failed / spawn node ENOENTClaude Desktop 启动子进程失败。最常见原因是args里的路径写错或者用了相对路径。必须用绝对路径且指向dist/index.js而不是index.ts。如果报ENOENT说明node不在 Claude Desktop 能识别的 PATH 里可以把command改成node的绝对路径用which node查。还有一种情况是编译没成功dist目录压根不存在回去跑一遍npm run build。reading choices of undefined这个报错来自summarize_and_cite里completion.choices[0]。原因通常是 OpenAI 调用失败但没抛异常返回体结构不对。检查OPENAI_API_KEY是否有效、额度是否够、模型名gpt-4o-mini是否拼对。另外max_tokens如果算出来是 0 或负数也会出问题代码里用了Math.min(Math.ceil(max_length / 4), 1000)max_length默认 500算出来是 125正常。如果你手动传了很小的max_length注意别传 0。OAuth / token 相关报错如果你在 Elasticsearch 侧用的是 OAuth token 而不是 API Keyauth字段的写法不一样要用auth: { bearer: token }。混用会导致认证失败。建议统一用 API Key简单直接。工具不出现或调用无响应先确认 Claude Desktop 完全重启了不是关窗口是退出进程再开。再看配置文件 JSON 有没有语法错误多一个逗号都会导致整个配置失效。如果工具列表里能看到但调用报错去 Claude Desktop 的日志目录看 stderr 输出MCP 服务器的console.log和异常都会打到那里。排查时有个通用思路先在终端手动跑一次node dist/index.js看它能不能正常启动不报错。如果手动跑就崩那问题在代码或环境变量如果手动跑正常但 Claude 里不行那问题在 Claude Desktop 的配置或路径。6. 把检索能力接进 AI 客户端后续可以怎么扩展跑通之后这套东西的扩展空间其实挺大。最直接的是加工具比如再加一个get_document_by_id让 Claude 能按 ID 精确取回某篇文档的全文或者加一个list_tags让它先看看索引里有哪些标签再决定怎么搜。工具多了之后Claude 会根据你的问题自动编排调用顺序你不需要在提示词里写先搜再总结。检索质量上search_docs里的 Query DSL 可以继续调。比如把multi_match的type改成best_fields或cross_fields针对不同字段组合效果不一样fuzziness从AUTO改成具体数字能控制容错强度再加一层filter按tags或时间范围过滤避免把过期文档喂给模型。这些改动都在一个函数里改完重新编译即可。如果你希望这套检索能力不只服务 Claude Desktop而是给更多编码场景或 Agent 用可以考虑把 MCP 服务器部署成一个长期运行的服务配合统一的模型接入层来管理 Key 和额度。TaoToken 提供了模型对话、Coding Plan、API Keys 和接入文档等入口适合把这类检索增强的 Agent 工作流沉淀下来长期使用。需要的话可以从 模型对话 先试一下模型调用或者直接看 接入文档 把 Base URL、Key、Model ID 三件套配好。长期跑编码类 Agent 的话Coding Plan 会更省心一些。最后一个实操建议把index.ts里的INDEX、字段权重、max_docs这些参数抽成环境变量这样同一份代码能接不同的索引不用每次改代码重编译。MCP 服务器本身很轻真正的价值在于你喂给它的检索逻辑和数据结构这部分值得多花点时间打磨。