|
| 1 | +from collections.abc import Callable |
| 2 | +from typing import Annotated |
| 3 | + |
1 | 4 | import mcp_types as types |
2 | 5 | import pytest |
| 6 | +from pydantic import Field |
3 | 7 |
|
4 | 8 | from mcp import Client |
5 | 9 | from mcp.server.mcpserver import Context, MCPServer |
| 10 | +from mcp.server.mcpserver.exceptions import InvalidSignature |
6 | 11 | from mcp.server.mcpserver.tools.base import Tool |
7 | 12 | from mcp.shared.exceptions import MCPError |
8 | 13 |
|
@@ -55,3 +60,77 @@ async def boom() -> str: |
55 | 60 |
|
56 | 61 | assert isinstance(result, types.CallToolResult) |
57 | 62 | assert result.is_error is True |
| 63 | + |
| 64 | + |
| 65 | +def array_header(tags: Annotated[list[str], Field(json_schema_extra={"x-mcp-header": "Tags"})]) -> str: |
| 66 | + raise NotImplementedError |
| 67 | + |
| 68 | + |
| 69 | +def number_header(ratio: Annotated[float, Field(json_schema_extra={"x-mcp-header": "Ratio"})]) -> str: |
| 70 | + raise NotImplementedError |
| 71 | + |
| 72 | + |
| 73 | +def optional_header(region: Annotated[str | None, Field(json_schema_extra={"x-mcp-header": "Region"})] = None) -> str: |
| 74 | + raise NotImplementedError |
| 75 | + |
| 76 | + |
| 77 | +def non_token_header(region: Annotated[str, Field(json_schema_extra={"x-mcp-header": "Region Name"})]) -> str: |
| 78 | + raise NotImplementedError |
| 79 | + |
| 80 | + |
| 81 | +@pytest.mark.anyio |
| 82 | +@pytest.mark.parametrize( |
| 83 | + ("fn", "reason"), |
| 84 | + [ |
| 85 | + pytest.param( |
| 86 | + array_header, |
| 87 | + "property 'tags': x-mcp-header is only permitted on integer/string/boolean properties (got 'array')", |
| 88 | + id="array", |
| 89 | + ), |
| 90 | + pytest.param( |
| 91 | + number_header, |
| 92 | + "property 'ratio': x-mcp-header is only permitted on integer/string/boolean properties (got 'number')", |
| 93 | + id="number", |
| 94 | + ), |
| 95 | + pytest.param( |
| 96 | + optional_header, |
| 97 | + "property 'region': x-mcp-header is only permitted on integer/string/boolean properties " |
| 98 | + "(the type keyword is NoneType, not a string)", |
| 99 | + id="optional", |
| 100 | + ), |
| 101 | + pytest.param( |
| 102 | + non_token_header, |
| 103 | + "property 'region': x-mcp-header 'Region Name' is not an RFC 9110 token", |
| 104 | + id="non-token-name", |
| 105 | + ), |
| 106 | + ], |
| 107 | +) |
| 108 | +async def test_tool_with_an_invalid_x_mcp_header_annotation_is_rejected_at_registration( |
| 109 | + fn: Callable[..., str], reason: str |
| 110 | +): |
| 111 | + """SDK-defined: the spec has 2026-07-28 clients exclude such a tool, so registration |
| 112 | + refuses it with an error naming the tool and the reason, and nothing is registered.""" |
| 113 | + mcp = MCPServer(name="srv") |
| 114 | + |
| 115 | + with pytest.raises(InvalidSignature) as exc_info: |
| 116 | + mcp.add_tool(fn) |
| 117 | + |
| 118 | + assert str(exc_info.value) == f"Tool {fn.__name__!r} has an invalid x-mcp-header annotation: {reason}" |
| 119 | + assert await mcp.list_tools() == [] |
| 120 | + |
| 121 | + |
| 122 | +@pytest.mark.anyio |
| 123 | +async def test_tool_with_valid_x_mcp_header_annotations_is_registered(): |
| 124 | + """SDK-defined: string, integer and boolean parameters may carry `x-mcp-header`, |
| 125 | + so a tool that annotates one of each registers.""" |
| 126 | + mcp = MCPServer(name="srv") |
| 127 | + |
| 128 | + @mcp.tool() |
| 129 | + def fetch( |
| 130 | + region: Annotated[str, Field(json_schema_extra={"x-mcp-header": "Region"})], |
| 131 | + shard: Annotated[int, Field(json_schema_extra={"x-mcp-header": "Shard"})], |
| 132 | + dry_run: Annotated[bool, Field(json_schema_extra={"x-mcp-header": "Dry-Run"})], |
| 133 | + ) -> str: |
| 134 | + raise NotImplementedError |
| 135 | + |
| 136 | + assert [tool.name for tool in await mcp.list_tools()] == ["fetch"] |
0 commit comments