Memory MCP Server

记忆 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 中实现更改