Reject a tool with an invalid x-mcp-header annotation at registration

src/mcp/server/mcpserver/tools/base.py

@@ -24,6 +24,7 @@
 from mcp.server.mcpserver.utilities.func_metadata import FuncMetadata, func_metadata
 from mcp.shared._callable_inspection import is_async_callable
 from mcp.shared.exceptions import MCPError
+from mcp.shared.inbound import find_invalid_x_mcp_header
 from mcp.shared.tool_name_validation import validate_and_warn_tool_name
 
 if TYPE_CHECKING:
@@ -104,6 +105,8 @@ def from_function(
             structured_output=structured_output,
         )
         parameters = func_arg_metadata.arg_model.model_json_schema(by_alias=True)
+        if (reason := find_invalid_x_mcp_header(parameters)) is not None:
+            raise InvalidSignature(f"Tool {func_name!r} has an invalid x-mcp-header annotation: {reason}")
 
         # Match `model_dump_one_level`'s kwarg keys (alias when present, else field name)
         # so a by-name resolver param resolves to a key that exists at call time.

tests/server/mcpserver/tools/test_base.py

@@ -1,8 +1,13 @@
+from collections.abc import Callable
+from typing import Annotated
+
 import mcp_types as types
 import pytest
+from pydantic import Field
 
 from mcp import Client
 from mcp.server.mcpserver import Context, MCPServer
+from mcp.server.mcpserver.exceptions import InvalidSignature
 from mcp.server.mcpserver.tools.base import Tool
 from mcp.shared.exceptions import MCPError
 
@@ -55,3 +60,77 @@ async def boom() -> str:
 
     assert isinstance(result, types.CallToolResult)
     assert result.is_error is True
+
+
+def array_header(tags: Annotated[list[str], Field(json_schema_extra={"x-mcp-header": "Tags"})]) -> str:
+    raise NotImplementedError
+
+
+def number_header(ratio: Annotated[float, Field(json_schema_extra={"x-mcp-header": "Ratio"})]) -> str:
+    raise NotImplementedError
+
+
+def optional_header(region: Annotated[str | None, Field(json_schema_extra={"x-mcp-header": "Region"})] = None) -> str:
+    raise NotImplementedError
+
+
+def non_token_header(region: Annotated[str, Field(json_schema_extra={"x-mcp-header": "Region Name"})]) -> str:
+    raise NotImplementedError
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("fn", "reason"),
+    [
+        pytest.param(
+            array_header,
+            "property 'tags': x-mcp-header is only permitted on integer/string/boolean properties (got 'array')",
+            id="array",
+        ),
+        pytest.param(
+            number_header,
+            "property 'ratio': x-mcp-header is only permitted on integer/string/boolean properties (got 'number')",
+            id="number",
+        ),
+        pytest.param(
+            optional_header,
+            "property 'region': x-mcp-header is only permitted on integer/string/boolean properties "
+            "(the type keyword is NoneType, not a string)",
+            id="optional",
+        ),
+        pytest.param(
+            non_token_header,
+            "property 'region': x-mcp-header 'Region Name' is not an RFC 9110 token",
+            id="non-token-name",
+        ),
+    ],
+)
+async def test_tool_with_an_invalid_x_mcp_header_annotation_is_rejected_at_registration(
+    fn: Callable[..., str], reason: str
+):
+    """SDK-defined: the spec has 2026-07-28 clients exclude such a tool, so registration
+    refuses it with an error naming the tool and the reason, and nothing is registered."""
+    mcp = MCPServer(name="srv")
+
+    with pytest.raises(InvalidSignature) as exc_info:
+        mcp.add_tool(fn)
+
+    assert str(exc_info.value) == f"Tool {fn.__name__!r} has an invalid x-mcp-header annotation: {reason}"
+    assert await mcp.list_tools() == []
+
+
+@pytest.mark.anyio
+async def test_tool_with_valid_x_mcp_header_annotations_is_registered():
+    """SDK-defined: string, integer and boolean parameters may carry `x-mcp-header`,
+    so a tool that annotates one of each registers."""
+    mcp = MCPServer(name="srv")
+
+    @mcp.tool()
+    def fetch(
+        region: Annotated[str, Field(json_schema_extra={"x-mcp-header": "Region"})],
+        shard: Annotated[int, Field(json_schema_extra={"x-mcp-header": "Shard"})],
+        dry_run: Annotated[bool, Field(json_schema_extra={"x-mcp-header": "Dry-Run"})],
+    ) -> str:
+        raise NotImplementedError
+
+    assert [tool.name for tool in await mcp.list_tools()] == ["fetch"]