Merge pull request #99 from Serverless-Devs/fix/sanitize-non-identifier-tool-field-names

fix: sanitize non-identifier field names in MCP/OpenAPI tool schemas

Author: OhYee

agentrun/integration/utils/tool.py

@@ -41,6 +41,7 @@
 from pydantic import (
     AliasChoices,
     BaseModel,
+    ConfigDict,
     create_model,
     Field,
     ValidationError,
@@ -1396,6 +1397,14 @@ def _create_function_with_signature(
         args_schema, "__agentrun_argument_aliases__", {}
     )
     if alias_map:
+        existing_param_names = {p.name for p in parameters}
+        # 防御性 sanitize: alias 要落到 inspect.Parameter 上, 非法字符
+        # （如 ``x-access-id``）会触发 ValueError。当前 alias 仅由
+        # ``_maybe_add_body_alias`` 写入 "query", 但未来可能扩展。
+        # 若 alias 被 sanitize, 同时把 sanitized 名字加进 alias_map 指向同一
+        # canonical, 以便 _normalize_tool_arguments 在调用方使用签名暴露的
+        # sanitized 名字时也能正确翻译。
+        extra_alias_entries: Dict[str, str] = {}
         for alias, canonical in alias_map.items():
             canonical_field = args_schema.model_fields.get(canonical)
             alias_annotation = (
@@ -1408,14 +1417,29 @@ def _create_function_with_signature(
                 and alias_annotation is not None
             ):
                 alias_annotation = Optional[alias_annotation]
+            alias_name = (
+                alias
+                if alias.isidentifier()
+                else _sanitize_python_identifier(alias)
+            )
+            if alias_name != alias and alias_name not in alias_map:
+                extra_alias_entries[alias_name] = canonical
+            if alias_name in existing_param_names:
+                continue
+            existing_param_names.add(alias_name)
             parameters.append(
                 inspect.Parameter(
-                    alias,
+                    alias_name,
                     inspect.Parameter.KEYWORD_ONLY,
                     default=None,
                     annotation=alias_annotation,
                 )
             )
+        if extra_alias_entries:
+            # 合并到 args_schema 的 alias map (避免就地改动原 dict)
+            merged = dict(alias_map)
+            merged.update(extra_alias_entries)
+            setattr(args_schema, "__agentrun_argument_aliases__", merged)
 
     # 创建实际执行函数
     def impl(**kwargs):
@@ -1425,7 +1449,9 @@ def impl(**kwargs):
         if args_schema is not None:
             try:
                 parsed = args_schema(**normalized_kwargs)
-                payload = parsed.model_dump(mode="python", exclude_unset=True)
+                payload = parsed.model_dump(
+                    mode="python", exclude_unset=True, by_alias=True
+                )
             except ValidationError as exc:
                 raise ValueError(
                     f"Invalid arguments for tool '{tool_name}': {exc}"
@@ -1674,6 +1700,34 @@ def _build_openapi_schema(
     return schema, tuple(body_field_names), alias_map
 
 
+_PY_KEYWORDS: Set[str] = set()
+
+
+def _sanitize_python_identifier(name: str) -> str:
+    """将任意字符串转换为合法的 Python 标识符
+
+    用于把 JSON Schema 中含 ``-`` / ``.`` 等字符的字段名（例如 ``x-access-id``）
+    映射成 Pydantic / ``inspect.Parameter`` 都能接受的字段名。原始名通过 alias
+    继续保留在 JSON Schema 和实际调用中。
+    """
+    import keyword
+
+    if not _PY_KEYWORDS:
+        _PY_KEYWORDS.update(keyword.kwlist)
+
+    sanitized = re.sub(r"[^0-9a-zA-Z_]", "_", name)
+    sanitized = sanitized.lstrip("_")
+    if not sanitized:
+        sanitized = "field"
+    if sanitized[0].isdigit():
+        # 数字开头不是合法 Python 标识符; 又因为 Pydantic 不允许字段名以
+        # 下划线开头, 这里只能加字母前缀 "field_" 而不是直接补 "_".
+        sanitized = "field_" + sanitized
+    if sanitized in _PY_KEYWORDS:
+        sanitized = sanitized + "_"
+    return sanitized
+
+
 def _json_schema_to_pydantic(
     name: str,
     schema: Optional[Dict[str, Any]],
@@ -1688,40 +1742,74 @@ def _json_schema_to_pydantic(
 
     required_fields = set(schema.get("required", []))
     fields = {}
+    needs_populate_by_name = False
+    used_py_names: Set[str] = set()
 
     for field_name, field_schema in properties.items():
         if not isinstance(field_schema, dict):
             continue
 
+        # 把含非法字符（如 ``x-access-id``）或保留字（``class``）的字段名映射到
+        # 合法的 Python 标识符, 通过 alias 保留原名以便 JSON Schema 输出和
+        # 调用真实 MCP 工具时使用。
+        import keyword as _kw
+
+        if field_name.isidentifier() and not _kw.iskeyword(field_name):
+            py_name = field_name
+        else:
+            py_name = _sanitize_python_identifier(field_name)
+        if py_name in used_py_names:
+            suffix = 2
+            while f"{py_name}_{suffix}" in used_py_names:
+                suffix += 1
+            py_name = f"{py_name}_{suffix}"
+        used_py_names.add(py_name)
+        if py_name != field_name:
+            needs_populate_by_name = True
+
         # 映射类型
         field_type = _json_type_to_python(field_schema)
         description = field_schema.get("description", "")
         default = field_schema.get("default")
         aliases = field_schema.get("x-aliases")
         field_kwargs: Dict[str, Any] = {"description": description}
+
+        # 用 ``alias`` 同时作用于 JSON Schema 输出和 by_alias dump,
+        # 让 LLM/调用端看到的字段名仍是原始名（如 ``x-access-id``）。
+        if py_name != field_name:
+            field_kwargs["alias"] = field_name
         if aliases:
             if not isinstance(aliases, (list, tuple)):
                 aliases = [aliases]
-            field_kwargs["validation_alias"] = AliasChoices(
-                field_name, *aliases
-            )
+            alias_choices: List[str] = [field_name]
+            if py_name != field_name:
+                alias_choices.append(py_name)
+            for alias in aliases:
+                if alias and alias not in alias_choices:
+                    alias_choices.append(alias)
+            field_kwargs["validation_alias"] = AliasChoices(*alias_choices)
 
         # 构建字段定义
         if field_name in required_fields:
             # 必填字段
-            fields[field_name] = (field_type, Field(**field_kwargs))
+            fields[py_name] = (field_type, Field(**field_kwargs))
         else:
             # 可选字段
             from typing import Optional as TypingOptional
 
-            fields[field_name] = (
+            fields[py_name] = (
                 TypingOptional[field_type],
                 Field(default=default, **field_kwargs),
             )
 
     # 创建模型,清理名称
     model_name = re.sub(r"[^0-9a-zA-Z]", "", name.title())
-    return create_model(model_name or "Args", **fields)  # type: ignore
+    model_kwargs: Dict[str, Any] = {}
+    if needs_populate_by_name:
+        model_kwargs["__config__"] = ConfigDict(populate_by_name=True)
+    return create_model(  # type: ignore
+        model_name or "Args", **model_kwargs, **fields
+    )
 
 
 def _json_type_to_python(field_schema: Dict[str, Any]) -> type:

tests/unittests/integration/test_tool_utils.py

@@ -10,10 +10,14 @@
 import pytest
 
 from agentrun.integration.utils.tool import (
+    _build_tool_from_meta,
+    _create_function_with_signature,
     _extract_core_schema,
+    _json_schema_to_pydantic,
     _load_json,
     _merge_schema_dicts,
     _normalize_tool_arguments,
+    _sanitize_python_identifier,
     _to_dict,
     CommonToolSet,
     from_pydantic,
@@ -702,3 +706,168 @@ def test_get_schema_from_parameters(self):
         assert "name" in schema["properties"]
         assert "age" in schema["properties"]
         assert "name" in schema.get("required", [])
+
+
+class TestSanitizePythonIdentifier:
+    """测试字段名 sanitizer"""
+
+    def test_valid_identifier_unchanged(self):
+        assert _sanitize_python_identifier("normal_name") == "normal_name"
+
+    def test_hyphenated_name(self):
+        assert _sanitize_python_identifier("x-access-id") == "x_access_id"
+
+    def test_dotted_name(self):
+        assert _sanitize_python_identifier("a.b.c") == "a_b_c"
+
+    def test_leading_digit_prefixed(self):
+        assert _sanitize_python_identifier("123abc") == "field_123abc"
+
+    def test_keyword_suffixed(self):
+        assert _sanitize_python_identifier("class") == "class_"
+
+    def test_empty_string(self):
+        assert _sanitize_python_identifier("") == "field"
+
+    def test_only_invalid_chars(self):
+        assert _sanitize_python_identifier("---") == "field"
+
+
+class TestJsonSchemaToPydanticInvalidFieldNames:
+    """覆盖 _json_schema_to_pydantic 对非法 Python 标识符字段名的处理"""
+
+    def test_hyphenated_field_name_builds_model(self):
+        """字段名含 '-' 时不应抛错, 且 JSON Schema 仍以原名暴露"""
+        schema = {
+            "type": "object",
+            "properties": {
+                "x-access-id": {"type": "string", "description": "id"},
+            },
+            "required": ["x-access-id"],
+        }
+
+        model = _json_schema_to_pydantic("Args", schema)
+
+        assert model is not None
+        assert "x_access_id" in model.model_fields
+        json_schema = model.model_json_schema()
+        assert "x-access-id" in json_schema["properties"]
+        assert "x-access-id" in json_schema["required"]
+
+    def test_keyword_field_name_sanitized(self):
+        schema = {
+            "type": "object",
+            "properties": {
+                "class": {"type": "string", "description": "py keyword"},
+            },
+        }
+
+        model = _json_schema_to_pydantic("Args", schema)
+
+        assert model is not None
+        assert "class_" in model.model_fields
+        assert "class" in model.model_json_schema()["properties"]
+
+    def test_accepts_both_original_and_sanitized_name(self):
+        schema = {
+            "type": "object",
+            "properties": {
+                "x-access-id": {"type": "string"},
+            },
+            "required": ["x-access-id"],
+        }
+
+        model = _json_schema_to_pydantic("Args", schema)
+
+        # 原名: 通过 alias
+        m1 = model(**{"x-access-id": "v1"})
+        assert m1.model_dump(by_alias=True) == {"x-access-id": "v1"}
+        # 沙化名: 通过 populate_by_name
+        m2 = model(x_access_id="v2")
+        assert m2.model_dump(by_alias=True) == {"x-access-id": "v2"}
+
+
+class TestCreateFunctionWithSignatureAliasSanitization:
+    """覆盖 _create_function_with_signature 对非法 alias 名的防御处理"""
+
+    def test_alias_with_hyphen_sanitized(self):
+        """`__agentrun_argument_aliases__` 含非法标识符 alias 时不应崩溃"""
+        from pydantic import BaseModel as _BM
+
+        class _Args(_BM):
+            query: str
+
+        setattr(_Args, "__agentrun_argument_aliases__", {"x-alias": "query"})
+
+        toolset = MagicMock()
+        func = _create_function_with_signature("demo", _Args, toolset, None)
+
+        import inspect as _inspect
+
+        sig = _inspect.signature(func)
+        # 主字段保留, alias 被 sanitize
+        assert "query" in sig.parameters
+        assert "x_alias" in sig.parameters
+
+    def test_call_via_sanitized_alias_name_routes_to_canonical(self):
+        """用签名暴露的 sanitized alias 名调用时也应翻译到 canonical 字段
+
+        回归 Copilot review 提出的: 仅 sanitize 签名不够, 还要让
+        _normalize_tool_arguments 认识 sanitized alias.
+        """
+        from pydantic import BaseModel as _BM
+        from pydantic import Field as _Field
+
+        class _Args(_BM):
+            query: str = _Field()
+
+        setattr(_Args, "__agentrun_argument_aliases__", {"x-alias": "query"})
+
+        toolset = MagicMock()
+        toolset.call_tool = MagicMock(return_value={"ok": True})
+        func = _create_function_with_signature("demo", _Args, toolset, None)
+
+        # 用沙化后的 alias 名 (签名暴露的形式) 调用
+        func(x_alias="hello")
+
+        call_kwargs = toolset.call_tool.call_args.kwargs
+        assert call_kwargs["arguments"] == {"query": "hello"}
+        # 同时验证: alias_map 已被扩展, 包含 sanitized 形式
+        merged_map = getattr(_Args, "__agentrun_argument_aliases__")
+        assert merged_map.get("x_alias") == "query"
+        assert merged_map.get("x-alias") == "query"
+
+
+class TestBuildToolFromMetaInvalidFieldNames:
+    """覆盖 _build_tool_from_meta 完整链路 (回归 'x-access-id' 加载失败)"""
+
+    def test_mcp_input_schema_with_hyphen_field(self):
+        """模拟 MCP 工具元数据包含 'x-access-id' 入参时仍可成功构造 Tool"""
+        toolset = MagicMock()
+        toolset.call_tool = MagicMock(return_value={"status": "ok"})
+
+        meta = {
+            "name": "demo-tool",
+            "description": "demo",
+            "input_schema": {
+                "type": "object",
+                "properties": {
+                    "x-access-id": {
+                        "type": "string",
+                        "description": "id",
+                    },
+                    "value": {"type": "integer"},
+                },
+                "required": ["x-access-id"],
+            },
+        }
+
+        tool_obj = _build_tool_from_meta(toolset, meta, None)
+
+        assert tool_obj is not None
+        # 调用工具时, MCP 应收到原始字段名 'x-access-id'
+        tool_obj.func(**{"x-access-id": "abc", "value": 1})
+        toolset.call_tool.assert_called_once()
+        call_kwargs = toolset.call_tool.call_args.kwargs
+        assert call_kwargs["arguments"]["x-access-id"] == "abc"
+        assert call_kwargs["arguments"]["value"] == 1