UIengF commented on code in PR #368: URL: https://github.com/apache/hugegraph-ai/pull/368#discussion_r3563343306
########## hugegraph-mcp/pyproject.toml: ########## @@ -0,0 +1,60 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +[project] +name = "hugegraph-mcp" +version = "0.1.0" +description = "FastMCP server that exposes HugeGraph schema & Gremlin tools over MCP/STDIO" +readme = "README.md" +license = { text = "Apache-2.0" } +requires-python = ">=3.10" +authors = [ + { name = "Apache HugeGraph Contributors", email = "[email protected]" }, +] + +dependencies = [ + "fastmcp>=2.2.0", + "hugegraph-python-client", Review Comment: Fixed in `1833b36`. The MCP dependency is now `hugegraph-python-client>=1.7.0`. CI builds both client and MCP wheels from the same checkout, installs both explicitly into an isolated venv, asserts versions `1.7.0`/`0.1.0`, and verifies graphspace graph routing plus graphspace-scoped auth routing. The same isolated smoke passed locally. ########## hugegraph-mcp/hugegraph_mcp/tools/ingest_graph_data.py: ########## @@ -0,0 +1,1197 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""图数据导入 — 结构化 graph_data 校验和 legacy AI-backed 写入链路。 + +=== MCP V1 导入路径说明 === + +当前存在两条导入路径： + +1. **Public V1 路径（推荐）** + import_graph_data_tool(mode="ingest") 路由到 manage_graph_data() + → graph_data_to_change_plan() → 本地 Gremlin 写入。 + 此路径在 server.py:_import_graph_data() 中分发，不依赖 + HugeGraph-AI 服务。 + +2. **Legacy AI-backed 路径（兼容保留）** + ingest_graph_data() 真实实现为 ingest_graph_data_via_ai()， + 通过 HugeGraph-AI /graph-import HTTP 接口写入。 + 仅供需要 AI 辅助属性映射的内部/legacy 场景使用， + 不做为 MCP V1 公共工具的默认导入链路。 + +validate_graph_payload() 对 vertices/edges 做全面 schema 校验： +- label 是否存在于 live schema +- properties 字段是否在对应 label 中定义 +- 主键是否提供 +- 边端点是否可解析 +- 类型匹配 +""" + +import json +from copy import deepcopy +from typing import Any +from uuid import uuid4 + +from hugegraph_mcp.config import MCPConfig +from hugegraph_mcp.envelope import ErrorType, envelope_err, envelope_ok +from hugegraph_mcp.guard import Capability, guard +from hugegraph_mcp.hugegraph_ai_client import post +from hugegraph_mcp.tools.schema_utils import ( + edge_schema_endpoint_label as _edge_schema_endpoint_label, + normalized_schema_summary, + primary_key_names as _primary_key_names, + property_names as _property_names, + schema_payload as _schema_payload, +) +from hugegraph_mcp.tools.live_schema import fetch_live_schema_or_none + + +def _property_types(raw_schema: dict[str, Any]) -> dict[str, str]: + types: dict[str, str] = {} + for prop in raw_schema.get("propertykeys", []): + if not isinstance(prop, dict): + continue + name = prop.get("name") + data_type = prop.get("data_type") + if isinstance(name, str) and isinstance(data_type, str): + types[name] = data_type.upper() + return types + + +def _value_matches_type(value: Any, data_type: str) -> bool: + if value is None: + return True + if data_type in {"TEXT", "UUID"}: + return isinstance(value, str) + if data_type in {"INT", "LONG", "BYTE"}: + return isinstance(value, int) and not isinstance(value, bool) + if data_type in {"FLOAT", "DOUBLE"}: + return isinstance(value, (int, float)) and not isinstance(value, bool) + if data_type == "BOOLEAN": + return isinstance(value, bool) + if data_type in {"DATE", "BLOB"}: + return isinstance(value, str) + return True + + +def _indexed_labels(raw_schema: dict[str, Any]) -> dict[str, set[str]]: + indexed = {"VERTEX": set(), "EDGE": set()} + for index in raw_schema.get("indexlabels", []): + if not isinstance(index, dict): + continue + base_label = index.get("base_label") or index.get("baseLabel") + if not isinstance(base_label, str): + continue + base_type = str(index.get("base_type") or index.get("baseType") or "").upper() + if base_type in {"VERTEX", "VERTEX_LABEL"}: + indexed["VERTEX"].add(base_label) + elif base_type in {"EDGE", "EDGE_LABEL"}: + indexed["EDGE"].add(base_label) + return indexed + + +def _edge_endpoint(edge: dict[str, Any], endpoint: str) -> tuple[Any, Any]: + if endpoint == "source": + label = edge.get("source_label") or edge.get("outVLabel") + value = edge.get("source") if "source" in edge else edge.get("outV") + else: + label = edge.get("target_label") or edge.get("inVLabel") + value = edge.get("target") if "target" in edge else edge.get("inV") + return label, value + + +def _has_mixed_endpoint_forms(edge: dict[str, Any], endpoint: str) -> bool: + if endpoint == "source": + return "source" in edge and "outV" in edge + return "target" in edge and "inV" in edge + + +def _identity_value_present(value: Any) -> bool: + return value is not None and value != "" + + +def _format_endpoint_value(value: Any) -> str: + return repr(value) + + +def _endpoint_identities( + label: str, + value: Any, + schema_primary_keys: dict[str, list[str]], +) -> tuple[list[tuple[str, str, Any]], str | None]: + # 边端点支持两种写法：直接给后端 id，或按顶点主键给对象。 + # 返回多个候选身份，是为了兼容用户传入 "1:alice" 这类后端 id + # 和 {"name": "alice"} 这类主键对象两种输入。 + identities: list[tuple[str, str, Any]] = [] + primary_keys = schema_primary_keys.get(label, []) + + if isinstance(value, dict): + explicit_id = value.get("id") + if _identity_value_present(explicit_id): + identities.append((label, "id", explicit_id)) + if primary_keys: + missing = [ + pk + for pk in primary_keys + if pk not in value or not _identity_value_present(value.get(pk)) + ] + if missing: + if identities: + return identities, None + return identities, missing[0] + identities.append( + (label, "pk", tuple(value.get(pk) for pk in primary_keys)) + ) + return identities, None + + if _identity_value_present(value): + identities.append((label, "id", value)) + if len(primary_keys) == 1: + pk_value = value + if isinstance(value, str) and ":" in value: + pk_value = value.split(":", 1)[1] + identities.append((label, "pk", (pk_value,))) + return identities, None + + +def _schema_plan_summary(live_schema: dict[str, Any] | None) -> dict[str, Any] | None: + raw = _schema_payload(live_schema) + if raw is None: + return None + return { + "vertexlabels": raw.get("vertexlabels", []), + "edgelabels": raw.get("edgelabels", []), + "propertykeys": raw.get("propertykeys", []), + "indexlabels": raw.get("indexlabels", []), + } + + +def _canonical_json_key(value: Any) -> str: + return json.dumps(value, sort_keys=True, default=str, separators=(",", ":")) + + +def _normalize_value(value: Any) -> Any: + if isinstance(value, dict): + return { + key: _normalize_value(value[key]) + for key in sorted(value, key=lambda item: str(item)) + } + if isinstance(value, list): + return [_normalize_value(item) for item in value] + return value + + +def _vertex_sort_key( + vertex: Any, + schema_primary_keys: dict[str, list[str]], +) -> tuple[str, str]: + if not isinstance(vertex, dict): + return ("", _canonical_json_key(vertex)) + label = str(vertex.get("label") or "") + if _identity_value_present(vertex.get("id")): + identity = vertex.get("id") + else: + props = vertex.get("properties") + primary_keys = schema_primary_keys.get(label, []) + if isinstance(props, dict) and primary_keys: + identity = props.get(primary_keys[0]) + elif isinstance(props, dict) and props: + first_key = sorted(props, key=lambda item: str(item))[0] + identity = props.get(first_key) + else: + identity = None + return (label, _canonical_json_key(identity)) + + +def _edge_sort_key(edge: Any) -> tuple[str, str, str, str]: + if not isinstance(edge, dict): + return ("", "", "", _canonical_json_key(edge)) + source_label, source = _edge_endpoint(edge, "source") + target_label, target = _edge_endpoint(edge, "target") + return ( + str(edge.get("label") or ""), + str(source_label or ""), + str(target_label or ""), + _canonical_json_key( + { + "source": source, + "target": target, + "properties": edge.get("properties", {}), + } + ), + ) + + +def _normalize_graph_data( + graph_data: dict[str, Any], + schema_summary: dict[str, Any] | None, +) -> dict[str, Any]: + # plan_hash 需要对输入顺序不敏感：同一批顶点/边即使 JSON 数组顺序不同， + # 也应得到同一个 hash；但属性值、schema 主键等安全相关内容必须参与 hash。 + normalized = _normalize_value(graph_data) + if not isinstance(normalized, dict): + return normalized + + schema_primary_keys: dict[str, list[str]] = {} + if schema_summary: + for vertex_label in schema_summary.get("vertexlabels", []): + if isinstance(vertex_label, dict): + name = vertex_label.get("name") + primary_keys = vertex_label.get("primary_keys") + if isinstance(name, str) and isinstance(primary_keys, list): + schema_primary_keys[name] = primary_keys + + vertices = normalized.get("vertices") + if isinstance(vertices, list): + normalized["vertices"] = sorted( + vertices, + key=lambda vertex: _vertex_sort_key(vertex, schema_primary_keys), + ) + + edges = normalized.get("edges") + if isinstance(edges, list): + normalized["edges"] = sorted(edges, key=_edge_sort_key) + + return normalized + + +def _schema_vertex_info(raw_schema: dict[str, Any]) -> dict[str, dict[str, Any]]: + info: dict[str, dict[str, Any]] = {} + for vertex_label in raw_schema.get("vertexlabels", []): + if not isinstance(vertex_label, dict): + continue + name = vertex_label.get("name") + if isinstance(name, str): + info[name] = { + "id": vertex_label.get("id"), + "primary_keys": _primary_key_names(vertex_label), + } + return info + + +def _canonical_primary_key_id( + label: str, + values: tuple[Any, ...], + vertex_info: dict[str, dict[str, Any]], +) -> str | None: + label_id = vertex_info.get(label, {}).get("id") + if label_id is None: + return None + return f"{label_id}:{'!'.join(str(value) for value in values)}" + + +def _vertex_backend_id( + vertex: dict[str, Any], + vertex_info: dict[str, dict[str, Any]], +) -> Any: + # HugeGraph PRIMARY_KEY 顶点的后端 id 由 label id 和主键值拼接而成。 + # 在导入前补齐 id，能让边端点在同一批 payload 内稳定引用刚创建的顶点。 + explicit_id = vertex.get("id") + if _identity_value_present(explicit_id): + return explicit_id + + label = vertex.get("label") + props = vertex.get("properties") + if not isinstance(label, str) or not isinstance(props, dict): + return None + + primary_keys = vertex_info.get(label, {}).get("primary_keys", []) + if not primary_keys: + return None + if not all( + pk in props and _identity_value_present(props.get(pk)) for pk in primary_keys + ): + return None + + values = tuple(props.get(pk) for pk in primary_keys) + return _canonical_primary_key_id(label, values, vertex_info) + + +def _vertex_identity_map( + vertices: list[Any], + raw_schema: dict[str, Any], +) -> tuple[dict[tuple[str, str, Any], Any], dict[str, list[str]]]: + # 建立"用户可表达的身份"到 HugeGraph 后端 id 的映射。 + # 同一顶点可能同时拥有显式 id、PRIMARY_KEY 后端 id 和主键 tuple， + # 边端点解析时任意一种命中都应该指向同一个后端顶点。 + vertex_info = _schema_vertex_info(raw_schema) + schema_primary_keys = { + label: info.get("primary_keys", []) for label, info in vertex_info.items() + } + identities: dict[tuple[str, str, Any], Any] = {} + + for vertex in vertices: + if not isinstance(vertex, dict): + continue + label = vertex.get("label") + if not isinstance(label, str): + continue + + backend_id = _vertex_backend_id(vertex, vertex_info) + if _identity_value_present(backend_id): + vertex.setdefault("id", backend_id) + identities[(label, "id", backend_id)] = backend_id + + explicit_id = vertex.get("id") + if _identity_value_present(explicit_id): + identities[(label, "id", explicit_id)] = backend_id or explicit_id + + props = vertex.get("properties") + primary_keys = schema_primary_keys.get(label, []) + if isinstance(props, dict) and primary_keys: + if all( + pk in props and _identity_value_present(props.get(pk)) + for pk in primary_keys + ): + values = tuple(props.get(pk) for pk in primary_keys) + identities[(label, "pk", values)] = backend_id or explicit_id + + return identities, schema_primary_keys + + +def _endpoint_backend_id( + label: str, + value: Any, + identities: dict[tuple[str, str, Any], Any], + schema_primary_keys: dict[str, list[str]], + vertex_info: dict[str, dict[str, Any]], +) -> Any: + # 边端点优先解析为本批 payload 中的顶点，保证"先创建顶点再创建边" + # 的批量导入场景不需要用户提前知道 HugeGraph 后端 id。 + endpoint_identities, _missing_pk = _endpoint_identities( + label, + value, + schema_primary_keys, + ) + for identity in endpoint_identities: + if identity in identities: + return identities[identity] + + if isinstance(value, dict): + explicit_id = value.get("id") + if _identity_value_present(explicit_id): + return explicit_id + + primary_keys = schema_primary_keys.get(label, []) + if primary_keys and all( + pk in value and _identity_value_present(value.get(pk)) + for pk in primary_keys + ): + values = tuple(value.get(pk) for pk in primary_keys) + return _canonical_primary_key_id(label, values, vertex_info) + return None + + if _identity_value_present(value): + return value + return None + + +def _prepare_graph_import_data( + graph_data: dict[str, Any], + live_schema: dict[str, Any], +) -> dict[str, Any]: + # HugeGraph-AI 的 graph-import 接口需要 outV/inV/outVLabel/inVLabel。 + # MCP 对用户暴露更友好的 source/target，因此这里在写入前做一次格式转换。 + prepared = deepcopy(graph_data) + raw_schema = _schema_payload(live_schema) or {} + vertex_info = _schema_vertex_info(raw_schema) + vertices = prepared.get("vertices") or [] + edges = prepared.get("edges") or [] + identities, schema_primary_keys = _vertex_identity_map(vertices, raw_schema) + + for edge in edges: + if not isinstance(edge, dict): + continue + src_label, source = _edge_endpoint(edge, "source") + tgt_label, target = _edge_endpoint(edge, "target") + edge.setdefault("properties", {}) + + if isinstance(src_label, str): + source_id = _endpoint_backend_id( + src_label, + source, + identities, + schema_primary_keys, + vertex_info, + ) + if _identity_value_present(source_id): + edge["outV"] = source_id + edge.setdefault("outVLabel", src_label) + + if isinstance(tgt_label, str): + target_id = _endpoint_backend_id( + tgt_label, + target, + identities, + schema_primary_keys, + vertex_info, + ) + if _identity_value_present(target_id): + edge["inV"] = target_id + edge.setdefault("inVLabel", tgt_label) + + return prepared + + +def validate_graph_payload( + graph_data: Any, + live_schema: dict[str, Any] | None = None, +) -> dict[str, Any]: + """校验 graph_data (vertices/edges) 与 live schema 的兼容性。 + + 覆盖：label 存在性、properties 字段合法性、主键完整性、 + 边端点可解析性、类型匹配、重复检测、索引建议。 + """ + errors: list[str] = [] + warnings: list[str] = [] + + if not isinstance(graph_data, dict): + return { + "valid": False, + "errors": ["graph_data must be an object"], + "warnings": warnings, + } + + vertices = graph_data.get("vertices") + edges = graph_data.get("edges") + + if not isinstance(vertices, list): + errors.append("vertices must be a list") + if not isinstance(edges, list): + errors.append("edges must be a list") + + schema_vlabels: set[str] = set() + schema_props: dict[str, set[str]] = {} + schema_primary_keys: dict[str, list[str]] = {} + schema_property_types: dict[str, str] = {} + schema_elabels: dict[str, dict[str, Any]] = {} + schema_eprops: dict[str, set[str]] = {} + indexed_labels = {"VERTEX": set(), "EDGE": set()} + raw = _schema_payload(live_schema) if live_schema is not None else None + schema_available = raw is not None + if schema_available: + # 把 live schema 摘成 label -> 属性/主键/类型表，后续校验只依赖这份快照。 + # 这避免遍历过程中 schema 被重复读取导致前后判断不一致。 + schema_property_types = _property_types(raw) + for vl in raw.get("vertexlabels", []): + if not isinstance(vl, dict): + continue + name = vl.get("name") + if name: + schema_vlabels.add(name) + schema_props[name] = _property_names(vl.get("properties", [])) + schema_primary_keys[name] = _primary_key_names(vl) + for el in raw.get("edgelabels", []): + if not isinstance(el, dict): + continue + name = el.get("name") + if isinstance(name, str): + schema_elabels[name] = el + schema_eprops[name] = _property_names(el.get("properties", [])) + indexed_labels = _indexed_labels(raw) + + vertex_labels: set[str] = set() + vertex_identity_index: dict[tuple[str, str, Any], int] = {} + if isinstance(vertices, list): + for idx, vertex in enumerate(vertices): + if not isinstance(vertex, dict): + errors.append(f"vertex {idx} must be an object") + continue + label = vertex.get("label") + if label in (None, ""): + errors.append(f"vertex {idx} missing required field: label") + continue + vertex_labels.add(label) + if schema_available and label not in schema_vlabels: + errors.append(f"vertex {idx} label '{label}' does not exist in schema") + + props = vertex.get("properties") + if isinstance(props, dict): + schema_prop_names = schema_props.get(label, set()) + for prop_name, prop_value in props.items(): + if prop_value is None or prop_value == "": + warnings.append( + f"vertex {idx} property '{prop_name}' has empty value" + ) + if ( + schema_available + and label in schema_props + and prop_name not in schema_prop_names + ): + errors.append( + f"vertex {idx} property '{prop_name}' does not exist on label '{label}'" + ) + data_type = schema_property_types.get(prop_name) + if data_type and not _value_matches_type(prop_value, data_type): + errors.append( + f"vertex {idx} property '{prop_name}' expects {data_type}, got {type(prop_value).__name__}" + ) + primary_keys = schema_primary_keys.get(label, []) Review Comment: Fixed in `1833b36`. Dry-run now reads `id_strategy`/`idStrategy` from the live schema and requires a non-empty string ID for `CUSTOMIZE_STRING` or a non-bool integer ID for `CUSTOMIZE_NUMBER`. Unit/public-entry tests cover missing and invalid IDs. Real HugeGraph 1.7.0 verification confirmed all four invalid cases are rejected during dry-run with zero writes. ########## hugegraph-mcp/hugegraph_mcp/tools/manage_schema.py: ########## @@ -0,0 +1,546 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Schema 管理统一入口 — design / validate / dry_run 三种模式。 + +V1 仅提供 schema 设计、校验和 dry-run 预览。公共 apply 工具保留 +FEATURE_DISABLED 响应，实际 schema 写入留到后续版本。 +""" + +import hashlib +import json +from copy import deepcopy +from typing import Any + +from hugegraph_mcp import schema_tools +from hugegraph_mcp.config import MCPConfig +from hugegraph_mcp.envelope import ErrorType, envelope_err, envelope_ok +from hugegraph_mcp.tools.live_schema import current_live_schema +from hugegraph_mcp.tools.schema_utils import normalized_schema_summary + + +ALLOWED_OPERATION_TYPES = frozenset( + { + "create_property_key", + "create_vertex_label", + "create_edge_label", + "create_index_label", + } +) + +REQUIRED_FIELDS = { + "create_property_key": ("name", "data_type"), + "create_vertex_label": ("name",), + "create_edge_label": ("name", "source_label", "target_label"), + "create_index_label": ("name", "base_type", "base_label"), +} + + +ValidationError = dict[str, Any] + + +def _operation_type(operation: dict[str, Any]) -> str: + return str(operation.get("type", "")) + + +def _is_delete_operation(op_type: str) -> bool: + lowered = op_type.lower() + return "delete" in lowered or "drop" in lowered + + +def _validation_error( + operation_index: int, + operation: Any, + reason: str, + suggestion: str, +) -> ValidationError: + return { + "operation_index": operation_index, + "operation": operation, + "reason": reason, + "suggestion": suggestion, + } + + +def _schema_items(live_schema: dict[str, Any], key: str) -> set[str]: + schema = live_schema.get("schema", {}) + return { + item.get("name") + for item in schema.get(key, []) + if isinstance(item, dict) and item.get("name") + } + + +def _collect_planned_creates( + operations: list[dict[str, Any]], +) -> tuple[dict[str, set[str]], list[ValidationError]]: + planned = { + "property_keys": set(), + "vertex_labels": set(), + "edge_labels": set(), + "index_labels": set(), + } + errors: list[ValidationError] = [] + create_type_to_key = { + "create_property_key": "property_keys", + "create_vertex_label": "vertex_labels", + "create_edge_label": "edge_labels", + "create_index_label": "index_labels", + } + create_type_to_label = { + "create_property_key": "property_key", + "create_vertex_label": "vertex_label", + "create_edge_label": "edge_label", + "create_index_label": "index_label", + } + + for idx, operation in enumerate(operations): + if not isinstance(operation, dict): + continue + + op_type = _operation_type(operation) + planned_key = create_type_to_key.get(op_type) + if planned_key is None: + continue + + name = operation.get("name") + if not name: + continue + + if name in planned[planned_key]: + errors.append( + _validation_error( + idx, + operation, + f"duplicate {op_type} name {name} within the same batch", + ( + f"Define each {create_type_to_label[op_type]} only once " + "per schema operation batch." + ), + ) + ) + continue + + planned[planned_key].add(name) + + return planned, errors + + +def _validate_property_references( + *, + idx: int, + operation: dict[str, Any], + field: str, + property_keys: set[str], + errors: list[ValidationError], +) -> None: + values = operation.get(field, []) + if values in (None, ""): + return + if not isinstance(values, list): + errors.append( + _validation_error( + idx, + operation, + f"{field} must be a list", + f"Use an array of existing property key names for {field}.", + ) + ) + return + + missing_properties = [name for name in values if name not in property_keys] + if missing_properties: + errors.append( + _validation_error( + idx, + operation, + f"{field} references undefined property key(s): {', '.join(missing_properties)}", + "Create these property keys first and rerun validation after they exist in the live schema.", + ) + ) + + +def _validation_warnings(operations: list[dict[str, Any]]) -> list[str]: + warnings: list[str] = [] + for idx, operation in enumerate(operations): + if not isinstance(operation, dict): + continue + if operation.get("type") == "create_vertex_label" and not operation.get( + "primary_keys" + ): + warnings.append( + f"operation {idx} (create_vertex_label) has no primary_keys definition" + ) + return warnings + + +def validate_schema_operations( + operations: list[dict[str, Any]], live_schema: dict[str, Any] | None = None +) -> dict[str, Any]: + """校验 schema 操作与 live schema 的兼容性。 + + 检查操作类型白名单、必填字段、property key 存在性、 + 边端点 label 存在性、索引 base_label 存在性、重复定义检测。 + """ + errors: list[ValidationError] = [] + + if not isinstance(operations, list): + return { + "valid": False, + "errors": [ + _validation_error( + -1, + operations, + "operations must be a list", + "Pass schema operations as a JSON array.", + ) + ], + "warnings": [], + } + + live_schema = current_live_schema(live_schema) + live_property_keys = _schema_items(live_schema, "propertykeys") + live_vertex_labels = _schema_items(live_schema, "vertexlabels") + live_edge_labels = _schema_items(live_schema, "edgelabels") + live_index_labels = _schema_items(live_schema, "indexlabels") + # 同一批 schema 操作允许前面的 create 被后续操作引用，例如先创建 + # property key，再创建使用它的 vertex label。planned_creates 用来模拟 + # 批内依赖，避免合法 migration 被误判为引用不存在。 + planned_creates, duplicate_errors = _collect_planned_creates(operations) + errors.extend(duplicate_errors) + + property_keys = live_property_keys | planned_creates["property_keys"] + vertex_labels = live_vertex_labels | planned_creates["vertex_labels"] + edge_labels = live_edge_labels | planned_creates["edge_labels"] + + for idx, operation in enumerate(operations): + if not isinstance(operation, dict): + errors.append( + _validation_error( + idx, + operation, + "operation must be an object", + "Replace this item with a schema operation object.", + ) + ) + continue + + op_type = _operation_type(operation) + if _is_delete_operation(op_type): + errors.append( + _validation_error( + idx, + operation, + f"unsupported delete/drop type: {op_type}", + "Use create-only schema operations; destructive schema changes are not supported.", + ) + ) + continue + + if op_type not in ALLOWED_OPERATION_TYPES: + errors.append( + _validation_error( + idx, + operation, + f"unsupported type: {op_type}", + "Use one of: create_property_key, create_vertex_label, create_edge_label, create_index_label.", + ) + ) + continue + + for field in REQUIRED_FIELDS[op_type]: Review Comment: Fixed across the hardened schema validation in the fourth commit, completed in `1833b36`. Validate/dry-run rejects invalid data type/cardinality/id strategy, invalid primary/nullable/sort/property references, and now explicitly rejects all supported aliases of parent/sub edge-label fields instead of silently ignoring them. Focused schema tests and the full MCP suite pass. ########## hugegraph-mcp/hugegraph_mcp/server.py: ########## @@ -0,0 +1,447 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""FastMCP 服务器入口 — MCP 工具注册和轻量 mode 路由。 + +每个 @mcp.tool() 装饰的函数就是一个对外暴露的 MCP 工具。 +server.py 只负责参数校验和 mode 分发，具体业务逻辑委托给 tools/ 下的模块。 +""" + +import logging +import logging.handlers +import os +import time +from typing import Any + +# ---- 启动时 patch：阻止 pyhugegraph 模块级日志初始化写入文件 ---- +# pyhugegraph 在 import 时会创建 RotatingFileHandler 写入 'logs/' 目录， +# 在 MCP stdio 模式下这会破坏 JSON 协议流，因此拦截 makedirs 和 RotatingFileHandler。 + +_original_makedirs = os.makedirs + + +def _safe_makedirs(name, mode=0o777, exist_ok=False): + if _is_logs_dir(name): + return None + return _original_makedirs(name, mode, exist_ok) + + +def _is_logs_dir(name) -> bool: + try: + path = os.fspath(name) + except TypeError: + return False + return os.path.basename(os.path.normpath(path)).lower() == "logs" + + +_OriginalRotatingFileHandler = logging.handlers.RotatingFileHandler + + +class _NoOpFileHandler(logging.NullHandler): + """无操作日志处理器 — 用于禁用文件日志记录。""" + + def __init__(self, *args, **kwargs): + super().__init__() + + +def _patched_rotating_handler(filename, *args, **kwargs): + if _is_logs_file(filename): + return _NoOpFileHandler() + return _OriginalRotatingFileHandler(filename, *args, **kwargs) + + +def _is_logs_file(filename) -> bool: + try: + path = os.path.normpath(os.fspath(filename)) + except TypeError: + return False + return any(part.lower() == "logs" for part in path.split(os.sep)) + + +logging.handlers.RotatingFileHandler = _patched_rotating_handler + +os.makedirs = _safe_makedirs + +try: + # ---- patch 作用域内，安全导入依赖 pyhugegraph 的模块 ---- + from fastmcp import FastMCP + + from hugegraph_mcp.config import MCPConfig + from hugegraph_mcp.envelope import ErrorType, envelope_err + from hugegraph_mcp.gremlin_tools import execute_gremlin_read, execute_gremlin_write + from hugegraph_mcp.guard import Capability + from hugegraph_mcp.tools.extract_graph_data import extract_graph_data + from hugegraph_mcp.tools.generate_gremlin import generate_gremlin + from hugegraph_mcp.tools.inspect_graph import inspect_graph + from hugegraph_mcp.tools.manage_graph_data import manage_graph_data + from hugegraph_mcp.tools.manage_schema import manage_schema + from hugegraph_mcp.tools.refresh_vid_embeddings import refresh_vid_embeddings +finally: + os.makedirs = _original_makedirs + logging.handlers.RotatingFileHandler = _OriginalRotatingFileHandler + +READONLY = MCPConfig.from_env().is_readonly() + +mcp = FastMCP("HugeGraph MCP") + + +def _align_public_tool_envelope( + result: dict[str, Any], + *, + tool_name: str, + duration_ms: float, +) -> dict[str, Any]: + """Add public wrapper metadata without changing the inner tool payload.""" + aligned = dict(result) + meta = dict(aligned.get("meta") or {}) + meta.setdefault("duration_ms", duration_ms) + aligned["meta"] = meta + + if aligned.get("ok") is False and isinstance(aligned.get("error"), dict): + error = dict(aligned["error"]) + error["source"] = tool_name + aligned["error"] = error + + return aligned + + +def _call_public_tool(tool_name: str, func, *args, **kwargs) -> dict[str, Any]: + start = time.perf_counter() + try: + result = func(*args, **kwargs) + except Exception as exc: + return envelope_err( + ErrorType.FLOW_EXECUTION_FAILED, + f"{tool_name} failed: {exc!s}", + source=tool_name, + details={"tool": tool_name}, + duration_ms=(time.perf_counter() - start) * 1000.0, + ) + return _align_public_tool_envelope( + result, + tool_name=tool_name, + duration_ms=(time.perf_counter() - start) * 1000.0, + ) + + +def _is_admin_mode_enabled() -> bool: + return MCPConfig.from_env().admin_mode + + +def _admin_gate(tool_name: str, *, requires_write: bool = False) -> dict | None: + """Return FEATURE_DISABLED envelope if admin mode is not enabled, else None.""" + if not _is_admin_mode_enabled(): + enable_env = {"admin_mode": "HUGEGRAPH_MCP_ADMIN_MODE"} + suggestion = f"Set HUGEGRAPH_MCP_ADMIN_MODE=true to enable {tool_name}." + if requires_write: + enable_env["readonly"] = "HUGEGRAPH_MCP_READONLY" + suggestion = ( + f"Set HUGEGRAPH_MCP_ADMIN_MODE=true and HUGEGRAPH_MCP_READONLY=false " + f"to enable {tool_name}." + ) + return envelope_err( + ErrorType.FEATURE_DISABLED, + f"{tool_name} is disabled by default in V1. Enable with HUGEGRAPH_MCP_ADMIN_MODE=true.", + suggestion=suggestion, + source=tool_name, + details={"tool": tool_name, "enable_env": enable_env}, + ) + + if requires_write and MCPConfig.from_env().is_readonly(): + return envelope_err( + ErrorType.READONLY_VIOLATION, + f"{tool_name} requires HUGEGRAPH_MCP_READONLY=false.", + suggestion=( + "Set HUGEGRAPH_MCP_ADMIN_MODE=true and HUGEGRAPH_MCP_READONLY=false " + "before retrying this admin write tool." + ), + source=tool_name, + details={ + "tool": tool_name, + "required_env": { + "HUGEGRAPH_MCP_ADMIN_MODE": "true", + "HUGEGRAPH_MCP_READONLY": "false", + }, + }, + readonly=True, + ) + return None + + +# ========== 工具 1：检视图状态和 schema ========== + + [email protected]() +def inspect_graph_tool(include_raw_schema: bool = False) -> dict: + """检视 HugeGraph 服务器状态、schema 摘要、点边计数和 AI 状态。 + + 推荐作为连接后第一个调用的工具。 + """ + return _call_public_tool( + "inspect_graph_tool", + inspect_graph, + include_raw_schema=include_raw_schema, + ) + + +# ========== V1 稳定工具 ========== + + [email protected]() +def generate_gremlin_tool( + query: str, + execute: bool = False, + output_types: list[str] | None = None, +) -> dict: + """V1 稳定工具：自然语言 → Gremlin 生成。 + + 默认不执行（execute=false），返回生成的 Gremlin 查询。 + 设置 execute=true 可执行生成的只读 Gremlin。 + """ + return _call_public_tool( + "generate_gremlin_tool", + generate_gremlin, + query=query, + execute=execute, + output_types=output_types, + ) + + [email protected]() +def execute_gremlin_read_tool(gremlin_query: str) -> dict: + """V1 稳定工具：执行只读 Gremlin 遍历查询。 + + 经过 GremlinPolicy 安全检查后执行。 + """ + return _call_public_tool( + "execute_gremlin_read_tool", + execute_gremlin_read, + gremlin_query, + ) + + [email protected]() +def extract_graph_data_tool( + text: str, + graph_schema: dict | None = None, + example_prompt: str | None = None, +) -> dict: + """V1 稳定工具：自然语言文本 → 候选 graph_data（不写入）。 + + 返回提取的顶点和边数据，供后续导入使用。 + graph_schema 可传入 HugeGraph schema；为空时使用当前图名作为 schema 引用。 + """ + return _call_public_tool( + "extract_graph_data_tool", + extract_graph_data, + text=text, + schema=graph_schema, + example_prompt=example_prompt, + ) + + [email protected]() +def design_schema_tool(operations: list[dict] | None = None) -> dict: + """V1 稳定工具：schema 设计指导。 + + 提供 schema 设计建议和最佳实践。 + """ + return _call_public_tool( + "design_schema_tool", + manage_schema, + mode="design", + operations=operations, + ) + + [email protected]() +def apply_schema_tool( + mode: str, + operations: list[dict] | None = None, + confirm: bool = False, + plan_hash: str | None = None, +) -> dict: + """V1 稳定工具：schema 校验和预览。 + + 支持 validate 和 dry_run 模式。apply 模式在 V1 中返回 FEATURE_DISABLED。 + """ + start = time.perf_counter() + if mode == "apply": + return envelope_err( + ErrorType.FEATURE_DISABLED, + "Schema apply is disabled in V1. Use validate or dry_run mode.", + suggestion="Use mode='validate' or mode='dry_run' to preview schema changes.", + source="apply_schema_tool", + details={"mode": mode, "tool": "apply_schema_tool"}, + duration_ms=(time.perf_counter() - start) * 1000.0, + ) + return _call_public_tool( + "apply_schema_tool", + manage_schema, + mode=mode, + operations=operations, + confirm=confirm, + plan_hash=plan_hash, + ) + + +# ========== 图数据导入入口 ========== + + [email protected]() +def import_graph_data_tool( + mode: str, + text: str | None = None, + graph_schema: dict | None = None, + example_prompt: str | None = None, + graph_data: dict | None = None, + table_data: dict | None = None, + mapping: dict | None = None, + dry_run: bool = True, + confirm: bool = False, + plan_hash: str | None = None, + nonce: str | None = None, + expires_at: float | None = None, +) -> dict: + """V1 图数据导入入口。 + + mode="extract": 自然语言文本 → 候选 graph_data + mode="ingest": MCP 本地校验+dry_run/confirm+Gremlin 导入 graph_data + mode="table": V1 禁用（返回 FEATURE_DISABLED） + """ + start = time.perf_counter() + + if mode == "extract": + if not text: + return envelope_err( + ErrorType.VALIDATION_ERROR, + "text is required for mode='extract'", + source="import_graph_data_tool", + duration_ms=(time.perf_counter() - start) * 1000.0, + ) + return _call_public_tool( + "import_graph_data_tool", + extract_graph_data, + text=text, + schema=graph_schema, + example_prompt=example_prompt, + ) + + if mode == "ingest": + if graph_data is None: + return envelope_err( + ErrorType.VALIDATION_ERROR, + "graph_data is required for mode='ingest'", + source="import_graph_data_tool", + duration_ms=(time.perf_counter() - start) * 1000.0, + ) + return _call_public_tool( + "import_graph_data_tool", + manage_graph_data, + mode="import", + graph_data=graph_data, + dry_run=dry_run, + confirm=confirm, + plan_hash=plan_hash, + nonce=nonce, + expires_at=expires_at, + plan_tool_name="import_graph_data_tool", + ) + + if mode == "table": + return envelope_err( + ErrorType.FEATURE_DISABLED, + "Table import is not available in V1.", + suggestion="Use mode='extract' with extract_graph_data_tool instead.", + source="import_graph_data_tool", + details={"mode": mode, "tool": "import_graph_data_tool"}, + duration_ms=(time.perf_counter() - start) * 1000.0, + ) + + return envelope_err( + ErrorType.VALIDATION_ERROR, + f"Unknown mode: {mode!r}. Use 'extract' or 'ingest'.", + source="import_graph_data_tool", + details={"mode": mode}, + duration_ms=(time.perf_counter() - start) * 1000.0, + ) + + +# ========== 受控图数据删除入口 ========== + + [email protected]() +def delete_graph_data_tool( + change_plan: dict, + dry_run: bool = True, + confirm: bool = False, + plan_hash: str | None = None, + nonce: str | None = None, + expires_at: float | None = None, +) -> dict: + """V1 稳定工具：受控删除图数据。 + + 只支持精确 delete_vertex/delete_edge change_plan。 + 必须经过 dry_run -> plan_hash -> confirm；不支持批量条件删除或级联删除。 + """ + return _call_public_tool( + "delete_graph_data_tool", + manage_graph_data, + mode="delete", + change_plan=change_plan, + dry_run=dry_run, + confirm=confirm, + plan_hash=plan_hash, + nonce=nonce, + expires_at=expires_at, + plan_tool_name="delete_graph_data_tool", + ) + + +# ========== 高级调试工具 ========== + + [email protected]() +def refresh_vid_embeddings_tool(confirm: bool = False) -> dict: + """手动刷新 VID 嵌入 — 需 admin mode 且 readonly=false。""" + blocked = _admin_gate("refresh_vid_embeddings_tool", requires_write=True) + if blocked: + return blocked + return _call_public_tool( + "refresh_vid_embeddings_tool", + refresh_vid_embeddings, + confirm=confirm, + ) + + [email protected]() Review Comment: Documented as an intentional break-glass exception in `1833b36`. The public English/Chinese safety contract and tool docstring now state that this admin-only tool has no preview, dry-run, plan hash, or confirmation and must only be enabled on an isolated trusted admin transport, never a shared agent/client endpoint. Its admin+readonly gate remains tested. ########## hugegraph-python-client/src/pyhugegraph/utils/log.py: ########## @@ -117,7 +117,19 @@ def init_logger( if rank > 0: log_filename = f"{log_filename}.rank{rank}" - os.makedirs(os.path.dirname(log_filename), exist_ok=True) + try: Review Comment: Fixed in `1833b36`. `init_logger()` now creates a directory only when `dirname` is non-empty, then continues to create the `RotatingFileHandler`. A real plain-filename test writes and reads `plain.log`; the client non-integration suite passes (`57 passed`). ########## hugegraph-mcp/hugegraph_mcp/hugegraph_ai_client.py: ########## @@ -0,0 +1,213 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""HugeGraph-AI HTTP 客户端 — 统一请求层。 + +所有 AI 调用经 request() 统一处理：allow_ai 开关检查、超时控制、 +Basic Auth 注入、结构化错误返回。不抛异常。 +""" + +import time +from typing import Any + +import requests + +from hugegraph_mcp.config import MCPConfig +from hugegraph_mcp.envelope import ErrorType, envelope_err, envelope_ok + + +def request( + method: str, + path: str, + *, + cfg: MCPConfig | None = None, + json: Any = None, + params: dict[str, Any] | None = None, + headers: dict[str, str] | None = None, +) -> dict[str, Any]: + """调用 HugeGraph-AI 并返回标准化信封。 + + allow_ai=False 时直接拒绝，连接超时/HTTP 错误/JSON 解析失败均返回 + HUGEGRAPH_AI_UNAVAILABLE 信封，不抛异常。 + """ + + start = time.perf_counter() + cfg = cfg or MCPConfig.from_env() + method = method.upper() + url = _build_url(cfg.ai_url, path) + + if not cfg.allow_ai: + return _ai_error( + "AI calls are disabled", + duration_ms=_duration_ms(start), + details={"method": method, "url": url}, + ) + + try: + kwargs: dict[str, Any] = { + "params": params, + "headers": headers, + "timeout": cfg.timeout_seconds, + } + if json is not None: + kwargs["json"] = json + if cfg.password: Review Comment: Fixed in `1833b36`. MCP now supports `HUGEGRAPH_AI_TOKEN` and injects `Authorization: Bearer ...`; an explicit Authorization header takes precedence case-insensitively. The incorrect HugeGraph graph password to AI Basic Auth coupling was removed. Config/client tests and README tables were updated. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
