Memgraph：高性能图数据库，兼容Neo4j，使用Cypher查询语言，专为实时图分析设计。
MCP：Model Context Protocol（模型上下文协议），AI世界的"USB-C"接口，实现AI模型与外部系统的标准化交互。
MCP-Memgraph集成：基于memgraph/ai-toolkit项目的实现，使AI模型能够通过标准化协议访问和操作Memgraph数据库。
Memgraph的MCP配置
原理、架构与设计思想
info介绍
architectureMCP架构与原理
Host
运行AI程序的环境，如Claude Desktop
Client
协议客户端，管理与服务器的连接
Server
MCP-Memgraph服务器，实现MCP协议
通信方式：stdio（标准输入输出）、SSE（Server-Sent Events）
核心构建块：工具（Tools）、资源（Resources）、提示（Prompts）
settingsMemgraph的MCP配置方法
环境准备：
# 克隆仓库
git clone https://github.com/memgraph/ai-toolkit.git
cd ai-toolkit/integrations/mcp-memgraph
# 安装依赖
pip install -e .
git clone https://github.com/memgraph/ai-toolkit.git
cd ai-toolkit/integrations/mcp-memgraph
# 安装依赖
pip install -e .
配置MCP客户端（Claude Desktop）：
{
"mcpServers": {
"memgraph": {
"command": "python",
"args": ["-m", "mcp_memgraph.server"],
"env": {
"MEMGRAPH_HOST": "localhost",
"MEMGRAPH_PORT": "7687",
"MEMGRAPH_USERNAME": "",
"MEMGRAPH_PASSWORD": "",
"MEMGRAPH_ENCRYPTED": "false"
}
}
}
}
"mcpServers": {
"memgraph": {
"command": "python",
"args": ["-m", "mcp_memgraph.server"],
"env": {
"MEMGRAPH_HOST": "localhost",
"MEMGRAPH_PORT": "7687",
"MEMGRAPH_USERNAME": "",
"MEMGRAPH_PASSWORD": "",
"MEMGRAPH_ENCRYPTED": "false"
}
}
}
}
启动Memgraph数据库：
docker run -it -p 7687:7687 -p 7444:7444 -p 3000:3000 memgraph/memgraph-platform
code使用示例和代码
核心代码结构：
# mcp_memgraph/server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from gqlalchemy import Memgraph
class MemgraphMCPServer:
def __init__(self):
self.server = Server("memgraph")
self.memgraph = None
self._setup_handlers()
def _setup_handlers(self):
@self.server.list_tools()
async def list_tools() -> list[dict]:
# 返回可用工具列表
pass
@self.server.call_tool()
async def call_tool(name: str, arguments: dict) -> dict:
# 处理工具调用
pass
from mcp.server import Server
from mcp.server.stdio import stdio_server
from gqlalchemy import Memgraph
class MemgraphMCPServer:
def __init__(self):
self.server = Server("memgraph")
self.memgraph = None
self._setup_handlers()
def _setup_handlers(self):
@self.server.list_tools()
async def list_tools() -> list[dict]:
# 返回可用工具列表
pass
@self.server.call_tool()
async def call_tool(name: str, arguments: dict) -> dict:
# 处理工具调用
pass
主要功能实现：
# 查询执行工具
@self.server.call_tool()
async def call_tool(name: str, arguments: dict) -> dict:
if name == "execute_query":
query = arguments.get("query", "")
try:
result = self.memgraph.execute_and_fetch(query)
return {"content": [{"type": "text", "text": str(list(result))}]}
except Exception as e:
return {"content": [{"type": "text", "text": f"Error: {str(e)}"}]}
# 获取图模式工具
elif name == "get_schema":
try:
schema = self._get_schema()
return {"content": [{"type": "text", "text": schema}]}
except Exception as e:
return {"content": [{"type": "text", "text": f"Error: {str(e)}"}]}
# 其他工具...
@self.server.call_tool()
async def call_tool(name: str, arguments: dict) -> dict:
if name == "execute_query":
query = arguments.get("query", "")
try:
result = self.memgraph.execute_and_fetch(query)
return {"content": [{"type": "text", "text": str(list(result))}]}
except Exception as e:
return {"content": [{"type": "text", "text": f"Error: {str(e)}"}]}
# 获取图模式工具
elif name == "get_schema":
try:
schema = self._get_schema()
return {"content": [{"type": "text", "text": schema}]}
except Exception as e:
return {"content": [{"type": "text", "text": f"Error: {str(e)}"}]}
# 其他工具...
实际使用示例：
# 启动服务器
if __name__ == "__main__":
server = MemgraphMCPServer()
stdio_server.run(server.server)
if __name__ == "__main__":
server = MemgraphMCPServer()
stdio_server.run(server.server)
tips_and_updates最佳实践和注意事项
-
security
安全考虑：使用环境变量存储敏感信息（如数据库密码），限制查询权限，防止注入攻击，使用参数化查询而非字符串拼接
-
speed
性能优化：实现查询结果缓存，限制返回数据量，避免大结果集，使用连接池管理数据库连接
-
bug_report
错误处理：提供清晰的错误信息，实现适当的日志记录，处理连接异常和查询错误
-
extension
扩展功能：添加更多图操作工具（如节点创建、关系更新），实现图算法调用接口，添加数据导入/导出功能