Neo4j MCP 服务器

这是一个内存控制协议 (MCP) 服务器的实现，它使用 Neo4j 作为后端存储来管理知识图谱。它提供了一个基于 stdio 的接口，用于以图数据库格式存储和检索知识。

前提条件

• Python 3.8+
• Neo4j 数据库（本地或远程）
• Poetry (Python 包管理器)
• Docker 和 Docker Compose（用于运行 Neo4j）
• Go Task（可选，用于任务自动化）

安装

1. 克隆存储库：

git clone <repository-url>
cd neo4j_mcp_server

2. 如果您尚未安装 Poetry，请安装它：

curl -sSL https://install.python-poetry.org | python3 -

3. 安装依赖项：

poetry install

配置

Claude Desktop 配置

对于运行 Claude Desktop 的 Ubuntu 用户，您可以通过将其添加到您的 Claude desktop 配置文件中来配置 MCP 服务器，该文件位于：

~/.config/Claude/claude_desktop_config.json

在配置之前，您需要构建独立的执行文件：

task build

这将在 dist/neo4j_mcp_server 创建一个二进制文件。请确保更新您的配置文件中的路径以指向此构建的执行文件。

example_mcp_config.json 中提供了一个示例配置。您可以复制并修改此文件：

cp example_mcp_config.json ~/.config/Claude/claude_desktop_config.json

然后编辑配置文件中的 command 路径以指向您构建的执行文件：

{
  "mcpServers": [
    {
      "name": "neo4j-knowledge-graph",
      "command": ["/path/to/your/dist/neo4j_mcp_server"],
      ...
    }
  ]
}

配置包括：

• 服务器名称和描述
• 启动服务器的命令（构建的执行文件的路径）
• 可用工具及其参数
• 必需字段和数据类型

运行服务器

使用 Task（推荐）

如果您安装了 Go Task，您可以使用提供的 Taskfile 来管理服务器：

# 显示可用任务
task

# 启动所有内容（Docker + 服务器）
task run

# 启动开发环境（Docker + 服务器 + 测试）
task dev

# 停止所有服务
task down

直接使用 Docker Compose

1. 启动 Neo4j 容器：

docker-compose up -d

2. 等待 Neo4j 准备就绪（容器将在 docker ps 中显示为“healthy”）

直接运行 MCP 服务器

使用以下命令启动服务器：

poetry run python mcp_neo4j_knowledge_graph/mcp/server.py

服务器将以 stdio 模式启动，准备好接受 MCP 协议消息。

可用工具

1. 创建实体

在知识图谱中创建新实体。每个实体必须具有类型和属性。如果未明确提供，则 ID 将自动从 name 属性设置。

参数：

• entities: 实体对象列表，每个对象包含：
  • type: 字符串 - 实体的类型（例如，Person，Organization）
  • properties: 对象 - 实体属性的键值对（必须包含 'id' 或 'name'）

示例输入：

{
    "entities": [{
        "type": "Person",
        "properties": {
            "name": "John Doe",
            "occupation": "Developer",
            "age": 30
        }
    }]
}

2. 创建关系

在知识图谱中创建现有实体之间的关系。在创建关系之前，所有引用的实体必须存在。

参数：

• relations: 关系对象列表，每个对象包含：
  • type: 字符串 - 关系的类型（例如，KNOWS，WORKS_FOR）
  • from: 字符串 - 源实体的 ID
  • to: 字符串 - 目标实体的 ID

示例输入：

{
    "relations": [{
        "type": "KNOWS",
        "from": "john_doe",
        "to": "jane_smith"
    }]
}

3. 搜索实体

在知识图谱中搜索实体，具有强大的文本匹配和过滤功能。可用于按文本搜索、按类型列出实体、查找具有特定属性的实体或这些过滤器的任意组合。

参数：

• search_term: 字符串（可选）- 要在实体属性中搜索的文本。如果未提供，则根据其他过滤器返回实体。
• entity_type: 字符串（可选）- 按实体类型过滤结果（例如，Person，Organization）。如果单独提供，则返回该类型的所有实体。
• properties: List[String]（可选）- 要过滤的属性名称列表：
  • 使用 search_term：在这些属性中搜索该术语
  • 不使用 search_term：返回具有任何这些已定义属性的实体
• include_relationships: 布尔值（可选，默认值：false）- 是否包括连接的实体和关系
• fuzzy_match: 布尔值（可选，默认值：true）- 当提供 search_term 时，是否使用不区分大小写的部分匹配

示例输入：

// 按文本搜索，带有类型过滤器
{
    "search_term": "John",
    "entity_type": "Person",
    "properties": ["name", "occupation"],
    "include_relationships": true
}

// 列出某个类型的所有实体
{
    "entity_type": "Person"
}

// 查找具有特定属性的实体
{
    "properties": ["email", "phone"],
    "entity_type": "Contact"
}

// 组合过滤器
{
    "entity_type": "Person",
    "properties": ["email"],
    "search_term": "example.com",
    "fuzzy_match": true
}

// 返回所有实体（无过滤器）
{}

返回：

{
    "results": [
        {
            "id": "john_doe",
            "type": ["Entity", "Person"],
            "properties": {
                "name": "John Doe",
                "email": "john@example.com"
            },
            "relationships": [  // 仅当 include_relationships 为 true 时才包含
                {
                    "type": "WORKS_AT",
                    "direction": "outgoing",
                    "node": {
                        "id": "tech_corp",
                        "type": "Company",
                        "properties": {
                            "name": "Tech Corp"
                        }
                    }
                }
            ]
        }
    ]
}

注意：

• 当未提供任何过滤器时，返回所有实体
• 实体类型过滤是精确匹配（不是模糊匹配）
• 属性存在性检查使用 IS NOT NULL 完成
• 文本搜索在 fuzzy_match 为 true 时支持不区分大小写的部分匹配
• 空结果作为空数组返回，而不是错误
• 性能考虑：
  • 按类型过滤比文本搜索更有效
  • 属性存在性检查已优化
  • 考虑使用特定属性而不是搜索所有属性
  • 大型结果集可能会在未来版本中分页

4. 更新实体

更新知识图谱中的现有实体。支持添加/删除属性和标签。

参数：

• updates: 更新对象列表，每个对象包含：
  • id: 字符串（必需）- 要更新的实体的 ID
  • properties: 对象（可选）- 要更新或添加的属性
  • remove_properties: List[String]（可选）- 要删除的属性名称
  • add_labels: List[String]（可选）- 要添加到实体的标签
  • remove_labels: List[String]（可选）- 要从实体中删除的标签

示例输入：

{
    "updates": [{
        "id": "john_doe",
        "properties": {
            "occupation": "Senior Developer",
            "salary": 100000
        },
        "remove_properties": ["temporary_note"],
        "add_labels": ["Verified"],
        "remove_labels": ["Pending"]
    }]
}

5. 删除实体

从知识图谱中删除实体，可以选择级联删除关系。

参数：

• entity_ids: List[String]（必需）- 要删除的实体 ID 列表
• cascade: 布尔值（可选，默认值：false）- 是否删除连接的关系
• dry_run: 布尔值（可选，默认值：false）- 预览删除影响而不进行更改

示例输入：

{
    "entity_ids": ["john_doe", "jane_smith"],
    "cascade": true,
    "dry_run": true
}

返回：

• success: 布尔值 - 操作是否成功
• deleted_entities: 已删除的实体列表
• deleted_relationships: 已删除的关系列表
• errors: 错误消息列表（如果有）
• impacted_entities: 将受影响的实体列表（仅限 dry_run）
• impacted_relationships: 将受影响的关系列表（仅限 dry_run）

6. 内省模式

检索有关 Neo4j 数据库模式的全面信息，包括节点标签、关系类型及其属性。

参数：不需要任何参数

返回：

• schema: 包含以下内容的对象：
  • node_labels: 数据库中所有节点标签的列表
  • relationship_types: 所有关系类型的列表
  • node_properties: 标签到属性名称列表的映射
  • relationship_properties: 关系类型到属性名称列表的映射

示例输入：

{}

测试

测试脚本

该项目包括几个用于系统不同方面的测试脚本：

1. mcp_neo4j_knowledge_graph/test_mcp_client.py - 测试 MCP 客户端功能

   • 验证服务器启动
   • 测试工具列表
   • 测试模式内省
   • 测试实体创建

   task test-client  # 仅运行客户端测试
   

2. mcp_neo4j_knowledge_graph/test_mcp_config.py - 测试 MCP 配置

   • 验证配置文件加载
   • 使用官方 MCP SDK 测试服务器连接
   • 验证所有必需的工具是否可用

   task test-config  # 仅运行配置测试
   

3. mcp_neo4j_knowledge_graph/test_neo4j_connection.py - 测试 Neo4j 数据库连接

   • 验证数据库连接
   • 测试基本查询功能
   • 检查环境配置

   task test-db  # 仅运行数据库测试
   

运行测试

您可以通过以下几种方式运行测试：

1. 一起运行所有测试：

   task test  # 运行所有测试，包括 pytest 和集成测试
   

2. 运行单个测试类型：

   task test-client    # 运行 MCP 客户端测试
   task test-config    # 运行 MCP 配置测试
   task test-db        # 运行 Neo4j 连接测试
   task test-integration  # 运行集成测试
   

3. 直接使用 pytest 运行测试：

   poetry run pytest  # 运行所有 pytest 兼容的测试
   

开发

使用 Task

该项目包括几个开发任务：

# 格式化代码
task format

# 运行 linter
task lint

# 运行测试
task test

# 启动开发环境
task dev

直接运行

该项目使用几个开发工具，这些工具会自动与 Poetry 一起安装：

• black 用于代码格式化
• isort 用于导入排序
• flake8 用于 linting
• pytest 用于测试

您可以使用 Poetry 运行这些工具：

# 格式化代码
poetry run black .

# 排序导入
poetry run isort .

# 运行 linter
poetry run flake8

# 运行测试
poetry run pytest

错误处理

服务器包括全面的错误处理，用于：

• 数据库连接问题
• 无效查询
• 缺少节点
• 无效的请求格式
• 模式验证错误
• 关系创建失败
• 实体更新冲突

所有错误都以 MCP 协议格式返回，并带有相应的错误消息。

Docker 配置

Neo4j 容器配置了以下设置：

• 端口：7474 (HTTP) 和 7687 (Bolt)
• 默认凭据：neo4j/password
• 启用 APOC 插件
• 启用文件导入/导出
• 配置了健康检查

您可以在 docker-compose.yml 文件中修改这些设置。

Task 命令参考

• task - 显示可用任务
• task run - 启动 Docker 和 MCP 服务器
• task dev - 启动开发环境（Docker + 服务器 + 测试）
• task docker - 启动 Neo4j 数据库
• task server - 运行 MCP 服务器
• task test - 运行所有测试
• task test-client - 运行 MCP 客户端测试
• task test-config - 运行 MCP 配置测试
• task test-db - 运行数据库测试
• task test-integration - 运行集成测试
• task down - 停止所有 Docker 服务
• task format - 使用 black 和 isort 格式化代码
• task lint - 运行 flake8 linter
• task help - 显示所有任务的详细帮助