Memory MCP 服务器

一个模型上下文协议 (MCP) 服务器，提供知识图谱功能，用于在内存中管理实体、关系和观察，并具有严格的验证规则以保持数据一致性。

安装

在 Claude Desktop 中安装服务器：

mcp install main.py -v MEMORY_FILE_PATH=/path/to/memory.jsonl

数据验证规则

实体名称

• 必须以小写字母开头
• 可以包含小写字母、数字和连字符
• 最大长度为 100 个字符
• 在图谱中必须是唯一的
• 有效名称示例：python-project、meeting-notes-2024、user-john

实体类型

支持以下实体类型：

• person: 人类实体
• concept: 抽象概念或原则
• project: 工作计划或任务
• document: 任何形式的文档
• tool: 软件工具或实用程序
• organization: 公司或团体
• location: 物理或虚拟地点
• event: 有时限的事件

观察

• 非空字符串
• 最大长度为 500 个字符
• 每个实体必须是唯一的
• 应该是事实和客观的陈述
• 在相关时包含时间戳

关系

支持以下关系类型：

• knows: 人与人之间的联系
• contains: 父/子关系
• uses: 实体利用另一个实体
• created: 作者/创建关系
• belongs-to: 成员/所有权
• depends-on: 依赖关系
• related-to: 通用关系

其他关系规则：

• 源实体和目标实体都必须存在
• 不允许自引用关系
• 不允许循环依赖
• 必须使用预定义的关系类型

用法

该服务器提供用于管理知识图谱的工具：

获取实体

result = await session.call_tool("get_entity", {
    "entity_name": "example"
})
if not result.success:
    if result.error_type == "NOT_FOUND":
        print(f"Entity not found: {result.error}")
    elif result.error_type == "VALIDATION_ERROR":
        print(f"Invalid input: {result.error}")
    else:
        print(f"Error: {result.error}")
else:
    entity = result.data
    print(f"Found entity: {entity}")

获取图谱

result = await session.call_tool("get_graph", {})
if result.success:
    graph = result.data
    print(f"Graph data: {graph}")
else:
    print(f"Error retrieving graph: {result.error}")

创建实体

# Valid entity creation
entities = [
    Entity(
        name="python-project",  # Lowercase with hyphens
        entityType="project",   # Must be a valid type
        observations=["Started development on 2024-01-29"]
    ),
    Entity(
        name="john-doe",
        entityType="person",
        observations=["Software engineer", "Joined team in 2024"]
    )
]
result = await session.call_tool("create_entities", {
    "entities": entities
})
if not result.success:
    if result.error_type == "VALIDATION_ERROR":
        print(f"Invalid entity data: {result.error}")
    else:
        print(f"Error creating entities: {result.error}")

添加观察

# Valid observation
result = await session.call_tool("add_observation", {
    "entity": "python-project",
    "observation": "Completed initial prototype"  # Must be unique for entity
})
if not result.success:
    if result.error_type == "NOT_FOUND":
        print(f"Entity not found: {result.error}")
    elif result.error_type == "VALIDATION_ERROR":
        print(f"Invalid observation: {result.error}")
    else:
        print(f"Error adding observation: {result.error}")

创建关系

# Valid relation
result = await session.call_tool("create_relation", {
    "from_entity": "john-doe",
    "to_entity": "python-project",
    "relation_type": "created"  # Must be a valid type
})
if not result.success:
    if result.error_type == "NOT_FOUND":
        print(f"Entity not found: {result.error}")
    elif result.error_type == "VALIDATION_ERROR":
        print(f"Invalid relation data: {result.error}")
    else:
        print(f"Error creating relation: {result.error}")

搜索内存

result = await session.call_tool("search_memory", {
    "query": "most recent workout"  # Supports natural language queries
})
if result.success:
    if result.error_type == "NO_RESULTS":
        print(f"No results found: {result.error}")
    else:
        results = result.data
        print(f"Search results: {results}")
else:
    print(f"Error searching memory: {result.error}")

搜索功能支持：

• 时间查询（例如，“最近的”，“最后的”，“最新的”）
• 活动查询（例如，“锻炼”，“运动”）
• 常规实体搜索
• 模糊匹配，相似度阈值为 80%
• 加权搜索：
  • 实体名称（权重：1.0）
  • 实体类型（权重：0.8）
  • 观察（权重：0.6）

删除实体

result = await session.call_tool("delete_entities", {
    "names": ["python-project", "john-doe"]
})
if not result.success:
    if result.error_type == "NOT_FOUND":
        print(f"Entity not found: {result.error}")
    else:
        print(f"Error deleting entities: {result.error}")

删除关系

result = await session.call_tool("delete_relation", {
    "from_entity": "john-doe",
    "to_entity": "python-project"
})
if not result.success:
    if result.error_type == "NOT_FOUND":
        print(f"Entity not found: {result.error}")
    else:
        print(f"Error deleting relation: {result.error}")

清空内存

result = await session.call_tool("flush_memory", {})
if not result.success:
    print(f"Error flushing memory: {result.error}")

错误类型

服务器使用以下错误类型：

• NOT_FOUND: 未找到实体或资源
• VALIDATION_ERROR: 无效的输入数据
• INTERNAL_ERROR: 服务器端错误
• ALREADY_EXISTS: 资源已存在
• INVALID_RELATION: 实体之间存在无效关系

响应模型

所有工具都使用以下模型返回类型化的响应：

EntityResponse

class EntityResponse(BaseModel):
    success: bool
    data: Optional[Dict[str, Any]] = None
    error: Optional[str] = None
    error_type: Optional[str] = None

GraphResponse

class GraphResponse(BaseModel):
    success: bool
    data: Optional[Dict[str, Any]] = None
    error: Optional[str] = None
    error_type: Optional[str] = None

OperationResponse

class OperationResponse(BaseModel):
    success: bool
    error: Optional[str] = None
    error_type: Optional[str] = None

开发

运行测试

pytest tests/

添加新功能

1. 更新 validation.py 中的验证规则
2. 在 tests/test_validation.py 中添加测试
3. 在 knowledge_graph_manager.py 中实现更改