Mirror x-mcp-header tool arguments into Mcp-Param-* request headers (SEP-2243) (#2990)

tests/shared/test_inbound.py

@@ -40,6 +40,8 @@ from mcp.shared.inbound import (
     decode_header_value,
     encode_header_value,
     find_invalid_x_mcp_header,
+    mcp_param_headers,
+    x_mcp_header_map,
 )
 
 CLIENT_INFO = {"name": "t", "version": "0"}
@@ -505,3 +507,70 @@ def test_find_invalid_x_mcp_header_reports_dotted_path_for_nested_property() ->
     schema = _schema(outer={"type": "object", "properties": {"r": {"type": "object", "x-mcp-header": "R"}}})
     reason = find_invalid_x_mcp_header(schema)
     assert reason is not None and "'outer.r'" in reason
+
+
+# --- x_mcp_header_map ----------------------------------------------------------
+
+
+def test_x_mcp_header_map_keys_top_level_and_nested_properties_by_path() -> None:
+    """Each annotated property maps to its token under its full `properties` path; unannotated props are absent."""
+    schema = _schema(
+        region={"type": "string", "x-mcp-header": "Region"},
+        query={"type": "string"},
+        outer={"type": "object", "properties": {"inner": {"type": "string", "x-mcp-header": "Inner"}}},
+    )
+    assert x_mcp_header_map(schema) == {("region",): "Region", ("outer", "inner"): "Inner"}
+
+
+@pytest.mark.parametrize("input_schema", [None, "not-a-mapping", {"type": "object"}])
+def test_x_mcp_header_map_empty_for_schemas_without_annotations(input_schema: Any) -> None:
+    assert x_mcp_header_map(input_schema) == {}
+
+
+# --- mcp_param_headers ---------------------------------------------------------
+
+
+def test_mcp_param_headers_renders_primitive_types_per_spec() -> None:
+    """String verbatim, integer as decimal, boolean as lowercase `true`/`false`, header named `Mcp-Param-<token>`."""
+    header_map = {("region",): "Region", ("priority",): "Priority", ("verbose",): "Verbose", ("debug",): "Debug"}
+    arguments = {"region": "us-west1", "priority": 42, "verbose": False, "debug": True}
+    assert mcp_param_headers(header_map, arguments) == {
+        "Mcp-Param-Region": "us-west1",
+        "Mcp-Param-Priority": "42",
+        "Mcp-Param-Verbose": "false",
+        "Mcp-Param-Debug": "true",
+    }
+
+
+@pytest.mark.parametrize(
+    ("value", "encoded"),
+    [
+        pytest.param("us-west1", "us-west1", id="plain-ascii"),
+        pytest.param("Hello, 世界", "=?base64?SGVsbG8sIOS4lueVjA==?=", id="non-ascii"),
+        pytest.param(" padded ", "=?base64?IHBhZGRlZCA=?=", id="edge-whitespace"),
+        pytest.param("line1\nline2", "=?base64?bGluZTEKbGluZTI=?=", id="control-char"),
+        pytest.param("=?base64?literal?=", "=?base64?PT9iYXNlNjQ/bGl0ZXJhbD89?=", id="sentinel-lookalike"),
+    ],
+)
+def test_mcp_param_headers_base64_wraps_header_unsafe_strings(value: str, encoded: str) -> None:
+    """Matches the spec's Value Encoding table: a non-header-safe string is base64-sentinel wrapped."""
+    assert mcp_param_headers({("v",): "Val"}, {"v": value}) == {"Mcp-Param-Val": encoded}
+
+
+def test_mcp_param_headers_omits_absent_or_null_arguments() -> None:
+    """A path that hits a missing key or a `None` value emits no header (spec: omit when no value is present)."""
+    header_map = {("present",): "Present", ("missing",): "Missing", ("nulled",): "Nulled"}
+    assert mcp_param_headers(header_map, {"present": "x", "nulled": None}) == {"Mcp-Param-Present": "x"}
+
+
+def test_mcp_param_headers_reads_nested_argument_path() -> None:
+    """A nested annotated property reads its value at the matching nested `arguments` path."""
+    headers = mcp_param_headers({("outer", "inner"): "Inner"}, {"outer": {"inner": "deep"}})
+    assert headers == {"Mcp-Param-Inner": "deep"}
+
+
+def test_mcp_param_headers_omits_when_nested_path_is_broken() -> None:
+    """A nested path through a non-mapping or missing intermediate node yields no header."""
+    header_map = {("outer", "inner"): "Inner"}
+    assert mcp_param_headers(header_map, {"outer": "not-a-mapping"}) == {}
+    assert mcp_param_headers(header_map, {}) == {}