博客

使用 Elasticsearch 构建 ChatGPT 连接器以查询 GitHub 问题

学习如何构建自定义 ChatGPT 连接器并部署使用混合搜索查询内部 GitHub 问题的 Elasticsearch MCP 服务器。

Tomás Murúa
智能体 AI混合搜索

Agent Builder is available now GA. Get started with an Elastic Cloud Trial, and check out the documentation for Agent Builder here.

最近，OpenAI 宣布为专业版/商务版/企业版和教育版 ChatGPT 提供自定义连接器功能。除了提供开箱即用的连接器来获取 Gmail、GitHub、Dropbox 等平台上的数据。还可以使用 MCP 服务器创建自定义连接器。

定制连接器使您能够将现有的 ChatGPT 连接器与其他数据源（如 Elasticsearch）结合，以获得全面的答案。

在本文中，我们将构建一个 MCP 服务器，将 ChatGPT 连接到包含内部 GitHub 问题和拉取请求信息的 Elasticsearch 索引。这样就可以使用 Elasticsearch 数据回答自然语言查询。

我们将在 Google Colab 上使用 FastMCP 和 ngrok 部署 MCP 服务器，以获取 ChatGPT 可以连接的公共 URL，从而省去复杂的基础架构设置。

有关 MCP 及其生态系统的全面概述，请参阅《MCP 的现状》。

准备工作

在开始之前，您需要：

  • Elasticsearch 集群（8.X 或更高版本）

  • Elasticsearch API密钥，具有对您的索引的读取访问权限

  • Google 账户（用于 Google Colab）

  • Ngrok账户 （免费套餐可用）

  • 拥有专业版/企业版/商务版或教育版套餐的 ChatGPT 账户

了解 ChatGPT MCP 连接器的要求

ChatGPT MCP 连接器需要实现两个工具：search 和 fetch。有关更多详情，请参阅 OpenAI 文档。

搜索工具

根据用户查询，从 Elasticsearch 索引中返回相关结果列表。

接收的内容：

  • 一个单一的字符串，包含用户的自然语言查询。

  • 示例：“查找与 Elasticsearch 迁移相关的问题。”

返回的内容：

  • 一个对象，其result 关键字包含一个结果对象数组。每个结果包括：

    • id - 唯一文档标识符

    • title - 问题或拉取请求标题

    • url - 链接到问题或 PR

在我们的实现中：

return {
    "results": [
        {
            "id": "PR-612",
            "title": "Fix memory leak in WebSocket notification service",
            "url": "https://internal-git.techcorp.com/pulls/612"
        },
        # ... more results
    ]
}

获取工具

获取指定文档的完整内容。

接收的内容：

  • 搜索结果中包含 Elasticsearch 文档 ID 的单个字符串

  • 示例：“获取 PR-578 的详细信息。”

它返回的内容：

  • 一个完整的文档对象，包含：

    • id - 唯一文档标识符

    • title - 问题或拉取请求标题

    • text - 完整的问题/PR描述和详细信息

    • url - 链接到问题或 PR

    • type - 文档类型（问题、pull_request）

    • status - 当前状态（打开、进行中、已解决）

    • priority - 优先级别（低、中、高、关键）

    • assignee - 负责此问题/PR 的人员

    • created_date - 何时创建

    • resolved_date - 何时解决（如适用）

    • labels - 与文件相关的标签

    • related_pr － 相关拉取请求 ID

return {
    "id": "PR-578",
    "title": "Security hotfix: Patch SQL injection vulnerabilities",
    "text": "Description: CRITICAL SECURITY FIX for ISSUE-1889. Patches SQL...",
    "url": "https://internal-git.techcorp.com/pulls/578",
    "type": "pull_request",
    "status": "closed",
    "priority": "critical",
    "assignee": "sarah_dev",
    "created_date": "2025-09-19",
    "resolved_date": "2025-09-19",
    "labels": "security, hotfix, sql",
    "related_pr": null
}

注意：本示例使用扁平结构，其中所有字段都位于根级别。OpenAI 的要求非常灵活，还支持嵌套的元数据对象。

GitHub 问题和 PR 数据集

在本教程中，我们将使用包含问题和拉取请求的内部 GitHub 数据集。这代表了一个您希望通过 ChatGPT 查询私有、内部数据的场景。

数据集可以在此处找到。我们将使用批量 API 更新数据索引。

这个数据集包含：

  • 有关描述、状态、优先级和分配人员的问题

  • 包含代码更改、审查和部署信息的拉取请求

  • 问题与 PR 之间的关系（例如，PR-578 修复了 ISSUE-1889）

  • 标签、日期和其他元数据

索引映射

该索引使用以下映射来支持使用 ELSER 的混合搜索。text_semantic 用于语义搜索，而其他字段用于关键字搜索。

{
  "mappings": {
    "properties": {
      "id": {
        "type": "keyword"
      },
      "title": {
        "type": "text"
      },
      "text": {
        "type": "text"
      },
      "text_semantic": {
        "type": "semantic_text",
        "inference_id": ".elser-2-elasticsearch"
      },
      "url": {
        "type": "keyword"
      },
      "type": {
        "type": "keyword"
      },
      "status": {
        "type": "keyword"
      },
      "priority": {
        "type": "keyword"
      },
      "assignee": {
        "type": "keyword"
      },
      "created_date": {
        "type": "date",
        "format": "iso8601"
      },
      "resolved_date": {
        "type": "date",
        "format": "iso8601"
      },
      "labels": {
        "type": "keyword"
      },
      "related_pr": {
        "type": "keyword"
      }
    }
  }
}

构建MCP服务器

我们的 MCP 服务器按照 OpenAI 规范实现了两个工具，使用混合搜索将语义和文本匹配相结合，以获得更好的结果。

搜索工具

利用 RRF（倒数排序融合）进行混合搜索，将语义搜索与文本匹配相结合：

@mcp.tool()
    async def search(query: str) -> Dict[str, List[Dict[str, Any]]]:
        """
        Search for internal issues and PRs using hybrid search (semantic + text with RRF).
        Returns list with id, title, and url per OpenAI spec.
        """
        if not query or not query.strip():
            return {"results": []}

        logger.info(f"Searching for: '{query}'")

        try:
            # Hybrid search with RRF (Reciprocal Rank Fusion)
            response = es_client.search(
                index=ELASTICSEARCH_INDEX,
                size=10,
                source=["id", "title", "url", "type", "priority"],
                retriever={
                    "rrf": {
                        "retrievers": [
                            {
                                # Semantic search with ELSER
                                "standard": {
                                    "query": {
                                        "semantic": {
                                            "field": "text_semantic",
                                            "query": query
                                        }
                                    }
                                }
                            },
                            {
                                # Text search (BM25) for keyword matching
                                "standard": {
                                    "query": {
                                        "multi_match": {
                                            "query": query,
                                            "fields": [
                                                "title^3",
                                                "text^2",
                                                "assignee^2",
                                                "type",
                                                "labels",
                                                "priority"
                                            ],
                                            "type": "best_fields",
                                            "fuzziness": "AUTO"
                                        }
                                    }
                                }
                            }
                        ],
                        "rank_window_size": 50,
                        "rank_constant": 60
                    }
                }
            )

            results = []
            if response and 'hits' in response:
                for hit in response['hits']['hits']:
                    source = hit['_source']
                    results.append({
                        "id": source.get('id', hit['_id']),
                        "title": source.get('title', 'Unknown'),
                        "url": source.get('url', '')
                    })

            logger.info(f"Found {len(results)} results")
            return {"results": results}

        except Exception as e:
            logger.error(f"Search error: {e}")
            raise ValueError(f"Search failed: {str(e)}")

要点：

  • 使用 RRF 的混合搜索：结合语义搜索 (ELSER) 和文本搜索 (BM25)，以获得更好的结果。

  • 多匹配查询：在多个字段中进行搜索，并使用增强功能（标题^3、文本^2、分配人员^2）。插入符号 (^) 会乘以相关性分数，优先考虑标题中的匹配项而非内容中的匹配项。

  • 模糊匹配：fuzziness: AUTO 通过允许近似匹配来处理错别字和拼写错误。

  • RRF 参数调整：

    • rank_window_size: 50 - 指定在合并之前从每个检索器（语义和文本）中考虑最靠前结果的数量。

    • rank_constant: 60 - 该值决定了单个结果集中的文档对最终排序结果的影响程度。

  • 仅返回必填字段：根据 OpenAI 规范返回 id、title、url，避免不必要地暴露其他字段。

获取工具

按文档 ID（如果存在）检索文档详细信息：

@mcp.tool()
    async def fetch(id: str) -> Dict[str, Any]:
        """
        Retrieve complete issue/PR details by ID.
        Returns id, title, text, url.
        """
        if not id:
            raise ValueError("ID is required")

        logger.info(f"Fetching: {id}")

        try:
            # Search by the 'id' field (not _id) since IDs are stored as a field
            response = es_client.search(
                index=ELASTICSEARCH_INDEX,
                body={
                    "query": {
                        "term": {
                            "id": id  # Search by your custom 'id' field
                        }
                    },
                    "size": 1
                }
            )

            if not response or not response['hits']['hits']:
                raise ValueError(f"Document with id '{id}' not found")

            hit = response['hits']['hits'][0]
            source = hit['_source']

            result = {
                "id": source.get('id', id),
                "title": source.get('title', 'Unknown'),
                "text": source.get('text', ''),
                "url": source.get('url', ''),
                "type": source.get('type', ''),
                "status": source.get('status', ''),
                "priority": source.get('priority', ''),
                "assignee": source.get('assignee', ''),
                "created_date": source.get('created_date', ''),
                "resolved_date": source.get('resolved_date', ''),
                "labels": source.get('labels', ''),
                "related_pr": source.get('related_pr', '')
            }

            logger.info(f"Fetched: {result['title']}")
            return result

        except Exception as e:
            logger.error(f"Fetch error: {e}")
            raise ValueError(f"Failed to fetch '{id}': {str(e)}")

要点：

  • 按文档 ID 字段进行搜索：使用自定义 id 字段上的术语查询

  • 返回完整文档：包含完整的 text 字段及其所有内容

  • 扁平结构：所有字段均位于根级别，与 Elasticsearch 的文档结构相匹配。

在 Google Colab 上部署

我们将使用 Google Colab 来运行 MCP 服务器，并使用 ngrok 将其公开，以便 ChatGPT 可以连接到它。

步骤 1：打开 Google Colab 笔记本

访问我们预配置的笔记本适用于 ChatGPT 的 Elasticsearch MCP。

步骤 2：配置您的凭据

您需要三项信息：

  • Elasticsearch URL：您的 Elasticsearch 集群 URL。

  • Elasticsearch API 密钥：具有索引读取权限的 API 密钥。

  • Ngrok 身份验证令牌：来自 ngrok 的免费令牌。我们将使用 ngrok 将 MCP URL 公开到互联网，以便 ChatGPT 可以连接到它。

获取 ngrok 令牌

  1. 在 ngrok 注册免费账户

  2. 前往您的 ngrok 仪表板

  3. 复制您的身份验证令牌

为 Google Colab 添加机密

在 Google Colab 笔记本中：

  1. 点击左侧边栏中的“密钥图标”以打开“机密”。

  2. 添加这三个秘密：

ELASTICSEARCH_URL=https://your-cluster.elastic.com:443
ELASTICSEARCH_API_KEY=your-api-key
NGROK_TOKEN=your-ngrok-token

3. 为每个机密启用笔记本访问权限

步骤 3：运行 Notebook

  1. 点击“运行时”，然后点击“全部运行”，以执行所有单元格

  2. 等待服务器启动（约30秒）

  3. 查找显示您的公开 ngrok URL 的输出

4. 该输出将显示如下内容：

连接 ChatGPT

现在我们将 MCP 服务器连接到您的 ChatGPT 账户。

  1. 打开 ChatGPT，前往“设置”。

  2. 导航到“连接器”。如果您使用的是专业版账户，则需要在连接器中打开“开发者模式”。

如果您使用的是 ChatGPT 企业版或商业版，您需要将连接器发布到您的工作场所。

3. 点击“创建”。

注意：在商业版、企业版和教育版工作区中，只有工作区所有者、管理员和已启用相应设置（针对企业版/教育版）的用户才能添加自定义连接器。具有普通成员角色的用户无法自行添加自定义连接器。

一旦连接器被所有者或管理员用户添加并启用，工作区中的所有成员即可使用该连接器。

4. 输入所需信息和以 /sse/ 结尾的 ngrok URL。请注意“sse”后面的“/”。没有它就无法正常工作：

  • 名字： Elasticsearch MCP

  • 描述：用于搜索和获取 GitHub 内部信息的自定义 MCP。

5. 按下“创建”保存自定义 MCP。

如果您的服务器正在运行，则连接是即时的。无需额外的身份验证，因为 Elasticsearch API 密钥已在服务器中配置。

测试 MCP 服务器

在提问之前，您需要先选择 ChatGPT 应该使用的连接器。

提示 1: 搜索问题

提问：“查找与 Elasticsearch 迁移相关的问题”并确认操作工具调用。

ChatGPT 将调用search 工具处理您的查询。你可以看到它正在查找可用工具，并准备调用 Elasticsearch 工具，在对该工具执行任何操作之前与用户确认。

工具调用请求：

{
  "query": "Elasticsearch migration issues"
}

工具响应：

{
  "results": [
    {
      "id": "PR-598",
      "title": "Elasticsearch 8.x migration - Application code changes",
      "url": "https://internal-git.techcorp.com/pulls/598"
    },
    {
      "id": "ISSUE-1712",
      "title": "Migrate from Elasticsearch 7.x to 8.x",
      "url": "https://internal-git.techcorp.com/issues/1712"
    },
    {
      "id": "RFC-045",
      "title": "Design Proposal: Microservices Migration Architecture",
      "url": "https://internal-git.techcorp.com/rfcs/045"
    }
    // ... 7 more results
  ]
}

ChatGPT 会处理这些结果，并以自然对话的形式呈现。

幕后

提示：“查找与 Elasticsearch 迁移相关的问题”

1. ChatGPT 调用 search(“Elasticsearch migration”)

2. Elasticsearch 执行混合搜索。

  • 语义搜索能理解“升级”和“版本兼容性”等概念。

  • 文本搜索可查找与“Elasticsearch”和“迁移”完全匹配的内容。

  • RRF 将两种方法的结果进行合并和排序

3. 返回与 id、title 匹配度最高的 10 个事件。 url

4. ChatGPT 将“ISSUE-1712：从 Elasticsearch 7.x 迁移到 8.x”作为最相关的结果

提示 2：获取完整的详细信息

问：“请提供有关 ISSUE-1889 的详细信息”

ChatGPT 识别到您需要有关特定问题的详细信息，并调用 fetch 工具，在对该工具采取任何行动前与用户确认。

工具调用请求：

{
  "id": "ISSUE-1889"
}

工具响应：

{
  "id": "ISSUE-1889",
  "title": "SQL injection vulnerability in search endpoint",
  "text": "Description: Security audit identified SQL injection vulnerability in /api/v1/search endpoint. User input from query parameter is not properly sanitized before being used in raw SQL query. Severity: HIGH - Immediate action required Affected Code: - File: services/search/query_builder.py - Line: 145-152 - Issue: String concatenation used instead of parameterized queries Investigation: - @security_team_alice: Confirmed exploitable with UNION-based injection - @sarah_dev: Checking all other endpoints for similar patterns - @john_backend: Found 3 more instances in legacy codebase Remediation: - Rewrite using SQLAlchemy ORM or parameterized queries - Add input validation and sanitization - Implement WAF rules as additional layer - Security regression tests Comments: - @tech_lead_mike: Stop all other work, this is P0 - @sarah_dev: PR-578 ready with fixes for all 4 vulnerable endpoints - @alex_devops: Deployed hotfix to production 2025-09-19 at 14:30 UTC - @security_team_alice: Verified fix, conducting full pentest next week Resolution: All vulnerable endpoints patched. Added pre-commit hooks to catch raw SQL queries. Security training scheduled for team.",
  "url": "https://internal-git.techcorp.com/issues/1889",
  "type": "issue",
  "status": "closed",
  "priority": "critical",
  "assignee": "sarah_dev",
  "created_date": "2025-09-18",
  "resolved_date": "2025-09-19",
  "labels": "security, vulnerability, bug, sql",
  "related_pr": "PR-578"
}

ChatGPT 会整合信息并清晰呈现。

幕后

提示：“获取有关 ISSUE-1889 的详细信息”

  1. ChatGPT 调用 fetch(“ISSUE-1889”)

  2. Elasticsearch 会检索完整文档

  3. 返回一个包含所有字段在根级别的完整文档

  4. ChatGPT会综合信息并提供正确的引用。

结论

在本文中，我们构建了一个自定义 MCP 服务器，使用专用的搜索和获取 MCP 工具将 ChatGPT 连接到 Elasticsearch，从而实现对私有数据的自然语言查询。

这种 MCP 模式适用于任何您想通过自然语言查询的 Elasticsearch 索引、文档、产品、日志或其他数据。

报告问题

相关内容

智能体 AI

如何使用 Mastra 和 Elasticsearch 构建代理式 AI 应用程序

Enrico Zimuel
智能体 AI集成

使用 TypeScript 构建 Elasticsearch MCP 服务器

Jeffrey Rengifo
智能体 AI

shell 工具并非上下文工程的灵丹妙药

Leonie Monigatti
智能体 AI集成

使用 Elasticsearch 推理 API 以及 Hugging Face 模型

Jeffrey Rengifo
集成智能体 AI

适用 Elasticsearch 的 Gemini CLI 扩展及工具和技能

Walter Rafelsberger