fix(mcp): emit correct field types in lean outputSchema (v1.33.18)

The lean outputSchema flattener defaulted every field to {"type":"object"}
whenever its type came from Pydantic's anyOf encoding of an optional
(T | None), because it only inspected a top-level "type" key. Strict MCP
clients then rejected valid tool responses whose values (strings, arrays,
numbers, booleans) did not match the advertised object type.

Resolve the real primitive from the non-null anyOf/oneOf arm and from a
one-hop $ref, while keeping the wire schema flat (no $defs/$ref/anyOf).
Fields with no single representable type (mixed-type unions or Any) now
emit a permissive empty schema that validates any value instead of a
wrong object type. A visited-set guard prevents unbounded recursion on
mutually-referential union definitions.

Adds field-type, edge-case, and cycle-guard tests. Closes #38.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: UPinar

app/config.py

@@ -12,7 +12,7 @@
 from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-VERSION = "1.33.17"
+VERSION = "1.33.18"
 MCP_TOOL_COUNT = 53  # v1.33.0: +tech_stack_cve_audit (MCP-only composite)
 MCP_RESOURCE_COUNT = 7  # v1.23.0: atlas+d3fend+cwe (4 templates + 3 catalogs)
 MCP_PROMPT_COUNT = 3  # v1.23.0: security_audit, vulnerability_check, contrast_triage

app/core/mcp_proxy.py

@@ -125,6 +125,48 @@ def _resolve_success(node: object) -> "dict | None":
             return defs.get(_ref.removeprefix("#/$defs/"))
         return node
 
+    def _ftype_of(_fs: object, _seen: "frozenset[str]" = frozenset()) -> dict:
+        # Resolve one field schema to a flat single-level wire fragment, usually
+        # {"type": <primitive>}. Handles a direct `type` (str or list form), a
+        # `$ref` (incl. ref->union, cycle-guarded) and Pydantic's `anyOf`/`oneOf`
+        # encoding of `T | None` / unions. Returns a permissive {} (validates any
+        # value) for fields with no single representable type — `Any` ({}) or a
+        # mixed-type union — so strict MCP clients never reject a valid response.
+        # Genuine complex models stay {"type": "object"}. `_seen` tracks visited
+        # $def names so a cyclic ref-of-unions can't recurse forever.
+        if not isinstance(_fs, dict):
+            return {"type": "object"}
+        if not _fs:
+            return {}
+        _t = _fs.get("type")
+        if isinstance(_t, str):
+            return {"type": _t}
+        if isinstance(_t, list):
+            _nn = [x for x in _t if isinstance(x, str) and x != "null"]
+            return {"type": _nn[0]} if len(_nn) == 1 else {}
+        _ref = _fs.get("$ref")
+        if isinstance(_ref, str) and _ref.startswith("#/$defs/"):
+            _name = _ref.removeprefix("#/$defs/")
+            _d = defs.get(_name) or {}
+            if isinstance(_d.get("type"), str):
+                return {"type": _d["type"]}
+            if (isinstance(_d.get("anyOf"), list) or isinstance(_d.get("oneOf"), list)) and _name not in _seen:
+                return _ftype_of(_d, _seen | {_name})
+            return {"type": "object"}
+        for _uk in ("anyOf", "oneOf"):
+            _arms = _fs.get(_uk)
+            if isinstance(_arms, list):
+                _frags = [_ftype_of(_a, _seen) for _a in _arms if isinstance(_a, dict) and _a.get("type") != "null"]
+                if not _frags:
+                    return {"type": "object"}
+                _types = {f.get("type") for f in _frags}
+                if len(_types) == 1 and None not in _types:
+                    return {"type": next(iter(_types))}
+                if len(_frags) == 1:
+                    return _frags[0]
+                return {}
+        return {"type": "object"}
+
     def _flatten(model: object) -> dict:
         # one-level: field name -> {"type": <primitive>}; nested obj/array kept as type only.
         if not isinstance(model, dict):
@@ -133,12 +175,7 @@ def _flatten(model: object) -> dict:
             return {"type": "array"}
         _props = {}
         for _fname, _fs in (model.get("properties") or {}).items():
-            _ftype = _fs.get("type") if isinstance(_fs, dict) else None
-            if isinstance(_fs, dict) and isinstance(_fs.get("$ref"), str) and _fs["$ref"].startswith("#/$defs/"):
-                _ftype = (defs.get(_fs["$ref"].removeprefix("#/$defs/")) or {}).get("type", "object")
-            if _ftype is None:
-                _ftype = "object"
-            _props[_fname] = {"type": _ftype}
+            _props[_fname] = _ftype_of(_fs)
         return {
             "type": "object",
             "properties": _props,

app/tests/test_mcp.py

@@ -1382,3 +1382,124 @@ def test_lean_outputschema_slowpath_cache_none(self, mcp_client, monkeypatch):
         assert r.status_code == 200
         assert r.json()["id"] == 99
         self._assert_lean_outputschema(r.json()["result"]["tools"])
+
+
+class TestLeanOutputSchemaFieldTypes:
+    """Issue #38: anyOf-encoded optional fields (Pydantic `T | None`) must keep
+    their real primitive type in the lean outputSchema, not collapse to
+    {"type": "object"}. Exercises _leanify_output_schema directly with a crafted
+    FastMCP-style envelope so the per-field type mapping is deterministic."""
+
+    _FASTMCP_OSCH = {
+        "$defs": {
+            "Verdict": {"type": "object", "properties": {"label": {"type": "string"}}},
+            "ErrorResponse": {"type": "object", "properties": {"error": {"type": "string"}}},
+            "CveSuccess": {
+                "type": "object",
+                "properties": {
+                    "summary": {"anyOf": [{"type": "string"}, {"type": "null"}]},
+                    "next_calls": {"anyOf": [{"type": "array", "items": {"type": "object"}}, {"type": "null"}]},
+                    "days_remaining": {"anyOf": [{"type": "integer"}, {"type": "null"}]},
+                    "cvss_v3": {"anyOf": [{"type": "number"}, {"type": "null"}]},
+                    "is_kev": {"anyOf": [{"type": "boolean"}, {"type": "null"}]},
+                    "verdict": {"anyOf": [{"$ref": "#/$defs/Verdict"}, {"type": "null"}]},
+                    "cve_id": {"type": "string"},
+                },
+                "required": ["cve_id"],
+            },
+        },
+        "properties": {"result": {"anyOf": [{"$ref": "#/$defs/CveSuccess"}, {"$ref": "#/$defs/ErrorResponse"}]}},
+        "required": ["result"],
+    }
+
+    def _leaned(self):
+        from core.mcp_proxy import _leanify_output_schema
+
+        out = _leanify_output_schema(self._FASTMCP_OSCH)
+        return out["properties"]["result"]["properties"]
+
+    def test_optional_string_keeps_string(self):
+        assert self._leaned()["summary"]["type"] == "string"
+
+    def test_optional_array_keeps_array(self):
+        assert self._leaned()["next_calls"]["type"] == "array"
+
+    def test_optional_integer_keeps_integer(self):
+        assert self._leaned()["days_remaining"]["type"] == "integer"
+
+    def test_optional_number_keeps_number(self):
+        assert self._leaned()["cvss_v3"]["type"] == "number"
+
+    def test_optional_boolean_keeps_boolean(self):
+        assert self._leaned()["is_kev"]["type"] == "boolean"
+
+    def test_optional_model_ref_stays_object(self):
+        assert self._leaned()["verdict"]["type"] == "object"
+
+    def test_plain_string_unaffected(self):
+        assert self._leaned()["cve_id"]["type"] == "string"
+
+    def test_schema_stays_flat(self):
+        import json as _json
+
+        from core.mcp_proxy import _leanify_output_schema
+
+        out = _leanify_output_schema(self._FASTMCP_OSCH)
+        blob = _json.dumps(out)
+        assert "$defs" not in out and "anyOf" not in blob and "$ref" not in blob
+
+
+class TestLeanOutputSchemaEdgeCases:
+    """Issue #38 hardening: shapes with no single representable flat type must
+    stay permissive ({} — validates any value) instead of a wrong
+    {"type": "object"}, and alternate encodings (type-as-list, $ref->union)
+    must still resolve to the real primitive."""
+
+    _OSCH = {
+        "$defs": {
+            "MaybeStr": {"anyOf": [{"type": "string"}, {"type": "null"}]},
+        },
+        "type": "object",
+        "properties": {
+            "type_list": {"type": ["string", "null"]},
+            "mixed_union": {"anyOf": [{"type": "integer"}, {"type": "string"}]},
+            "ref_to_union": {"$ref": "#/$defs/MaybeStr"},
+            "any_field": {},
+        },
+        "required": [],
+    }
+
+    def _leaned(self):
+        from core.mcp_proxy import _leanify_output_schema
+
+        return _leanify_output_schema(self._OSCH)["properties"]
+
+    def test_type_list_picks_non_null(self):
+        assert self._leaned()["type_list"] == {"type": "string"}
+
+    def test_mixed_union_is_permissive(self):
+        # cannot be a single flat type -> {} so strict clients never reject
+        assert self._leaned()["mixed_union"] == {}
+
+    def test_ref_to_union_resolves(self):
+        assert self._leaned()["ref_to_union"] == {"type": "string"}
+
+    def test_any_field_is_permissive(self):
+        assert self._leaned()["any_field"] == {}
+
+    def test_cyclic_ref_does_not_recurse_forever(self):
+        # A mutual $ref cycle between two union-defs must terminate (cycle guard),
+        # not blow the stack, and fall back to a safe wire fragment.
+        from core.mcp_proxy import _leanify_output_schema
+
+        osch = {
+            "$defs": {
+                "A": {"anyOf": [{"$ref": "#/$defs/B"}, {"type": "null"}]},
+                "B": {"anyOf": [{"$ref": "#/$defs/A"}, {"type": "null"}]},
+            },
+            "type": "object",
+            "properties": {"f": {"$ref": "#/$defs/A"}},
+            "required": [],
+        }
+        out = _leanify_output_schema(osch)
+        assert out["properties"]["f"] in ({"type": "object"}, {})