一个提供知识图谱功能的模型上下文协议(MCP)服务器,用于在内存中管理实体、关系和观察,并通过严格的验证规则来维护数据一致性。
在Claude Desktop上安装服务器:
mcp install main.py -v MEMORY_FILE_PATH=/path/to/memory.jsonl
python-project,meeting-notes-2024,user-john支持以下实体类型:
person:人类实体concept:抽象概念或原则project:工作计划或任务document:任何形式的文档tool:软件工具或实用程序organization:公司或团体location:物理或虚拟地点event:时间限定事件支持以下关系类型:
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"未找到实体: {result.error}")
elif result.error_type == "VALIDATION_ERROR":
print(f"无效输入: {result.error}")
else:
print(f"错误: {result.error}")
else:
entity = result.data
print(f"找到实体: {entity}")
result = await session.call_tool("get_graph", {})
if result.success:
graph = result.data
print(f"图数据: {graph}")
else:
print(f"检索图时出错: {result.error}")
# 有效的实体创建
entities = [
Entity(
name="python-project", # 小写并带有连字符
entityType="project", # 必须是有效的类型
observations=["2024-01-29 开始开发"]
),
Entity(
name="john-doe",
entityType="person",
observations=["软件工程师", "2024年加入团队"]
)
]
result = await session.call_tool("create_entities", {
"entities": entities
})
if not result.success:
if result.error_type == "VALIDATION_ERROR":
print(f"无效实体数据: {result.error}")
else:
print(f"创建实体时出错: {result.error}")
# 有效的观察
result = await session.call_tool("add_observation", {
"entity": "python-project",
"observation": "完成初步原型" # 对于实体必须是唯一的
})
if not result.success:
if result.error_type == "NOT_FOUND":
print(f"未找到实体: {result.error}")
elif result.error_type == "VALIDATION_ERROR":
print(f"无效观察: {result.error}")
else:
print(f"添加观察时出错: {result.error}")
# 有效的关系
result = await session.call_tool("create_relation", {
"from_entity": "john-doe",
"to_entity": "python-project",
"relation_type": "created" # 必须是有效的类型
})
if not result.success:
if result.error_type == "NOT_FOUND":
print(f"未找到实体: {result.error}")
elif result.error_type == "VALIDATION_ERROR":
print(f"无效关系数据: {result.error}")
else:
print(f"创建关系时出错: {result.error}")
result = await session.call_tool("search_memory", {
"query": "最近一次锻炼" # 支持自然语言查询
})
if result.success:
if result.error_type == "NO_RESULTS":
print(f"未找到结果: {result.error}")
else:
results = result.data
print(f"搜索结果: {results}")
else:
print(f"搜索记忆时出错: {result.error}")
搜索功能支持:
result = await session.call_tool("delete_entities", {
"names": ["python-project", "john-doe"]
})
if not result.success:
if result.error_type == "NOT_FOUND":
print(f"未找到实体: {result.error}")
else:
print(f"删除实体时出错: {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"未找到实体: {result.error}")
else:
print(f"删除关系时出错: {result.error}")
result = await session.call_tool("flush_memory", {})
if not result.success:
print(f"清除记忆时出错: {result.error}")
服务器使用以下错误类型:
NOT_FOUND:未找到实体或资源VALIDATION_ERROR:无效输入数据INTERNAL_ERROR:服务器内部错误ALREADY_EXISTS:资源已存在INVALID_RELATION:实体间的无效关系所有工具均使用这些模型返回类型化响应:
class EntityResponse(BaseModel):
success: bool
data: Optional[Dict[str, Any]] = None
error: Optional[str] = None
error_type: Optional[str] = None
class GraphResponse(BaseModel):
success: bool
data: Optional[Dict[str, Any]] = None
error: Optional[str] = None
error_type: Optional[str] = None
class OperationResponse(BaseModel):
success: bool
error: Optional[str] = None
error_type: Optional[str] = None
pytest tests/
validation.py中的验证规则tests/test_validation.py中添加测试knowledge_graph_manager.py中实现更改