|
1 | 1 | """Tests for issue #1574: Python SDK incorrectly validates Resource URIs. |
2 | 2 |
|
3 | | -The Python SDK uses Pydantic's AnyUrl for URI fields, which rejects relative paths |
4 | | -like 'users/me' that are valid according to the MCP spec and accepted by the |
5 | | -TypeScript SDK. |
| 3 | +The Python SDK previously used Pydantic's AnyUrl for URI fields, which rejected |
| 4 | +relative paths like 'users/me' that are valid according to the MCP spec and |
| 5 | +accepted by the TypeScript SDK. |
6 | 6 |
|
7 | | -The spec defines uri fields as plain strings with no JSON Schema format validation. |
8 | | -""" |
9 | | - |
10 | | -from mcp import types |
11 | | - |
12 | | - |
13 | | -class TestResourceUriValidation: |
14 | | - """Test that Resource URI fields accept all valid MCP URIs.""" |
15 | | - |
16 | | - def test_relative_path_uri(self): |
17 | | - """ |
18 | | - REPRODUCER: Relative paths like 'users/me' should be accepted. |
19 | | -
|
20 | | - Currently fails with: |
21 | | - ValidationError: Input should be a valid URL, relative URL without a base |
22 | | - """ |
23 | | - # This should NOT raise - relative paths are valid per MCP spec |
24 | | - resource = types.Resource(name="test", uri="users/me") |
25 | | - assert str(resource.uri) == "users/me" |
26 | | - |
27 | | - def test_custom_scheme_uri(self): |
28 | | - """Custom scheme URIs should be accepted.""" |
29 | | - resource = types.Resource(name="test", uri="custom://resource") |
30 | | - assert str(resource.uri) == "custom://resource" |
31 | | - |
32 | | - def test_file_url(self): |
33 | | - """File URLs should be accepted.""" |
34 | | - resource = types.Resource(name="test", uri="file:///path/to/file") |
35 | | - assert str(resource.uri) == "file:///path/to/file" |
36 | | - |
37 | | - def test_http_url(self): |
38 | | - """HTTP URLs should be accepted.""" |
39 | | - resource = types.Resource(name="test", uri="https://example.com/resource") |
40 | | - assert str(resource.uri) == "https://example.com/resource" |
41 | | - |
42 | | - |
43 | | -class TestReadResourceRequestParamsUri: |
44 | | - """Test that ReadResourceRequestParams.uri accepts all valid MCP URIs.""" |
45 | | - |
46 | | - def test_relative_path_uri(self): |
47 | | - """Relative paths should be accepted in read requests.""" |
48 | | - params = types.ReadResourceRequestParams(uri="users/me") |
49 | | - assert str(params.uri) == "users/me" |
50 | | - |
51 | | - |
52 | | -class TestResourceContentsUri: |
53 | | - """Test that ResourceContents.uri accepts all valid MCP URIs.""" |
54 | | - |
55 | | - def test_relative_path_uri(self): |
56 | | - """Relative paths should be accepted in resource contents.""" |
57 | | - contents = types.TextResourceContents(uri="users/me", text="content") |
58 | | - assert str(contents.uri) == "users/me" |
59 | | - |
60 | | - |
61 | | -class TestSubscribeRequestParamsUri: |
62 | | - """Test that SubscribeRequestParams.uri accepts all valid MCP URIs.""" |
63 | | - |
64 | | - def test_relative_path_uri(self): |
65 | | - """Relative paths should be accepted in subscribe requests.""" |
66 | | - params = types.SubscribeRequestParams(uri="users/me") |
67 | | - assert str(params.uri) == "users/me" |
68 | | - |
69 | | - |
70 | | -class TestUnsubscribeRequestParamsUri: |
71 | | - """Test that UnsubscribeRequestParams.uri accepts all valid MCP URIs.""" |
72 | | - |
73 | | - def test_relative_path_uri(self): |
74 | | - """Relative paths should be accepted in unsubscribe requests.""" |
75 | | - params = types.UnsubscribeRequestParams(uri="users/me") |
76 | | - assert str(params.uri) == "users/me" |
| 7 | +The fix changed URI fields to plain strings to match the spec, which defines |
| 8 | +uri fields as strings with no JSON Schema format validation. |
77 | 9 |
|
| 10 | +These tests verify the fix works end-to-end through the JSON-RPC protocol. |
| 11 | +""" |
78 | 12 |
|
79 | | -class TestResourceUpdatedNotificationParamsUri: |
80 | | - """Test that ResourceUpdatedNotificationParams.uri accepts all valid MCP URIs.""" |
| 13 | +import pytest |
81 | 14 |
|
82 | | - def test_relative_path_uri(self): |
83 | | - """Relative paths should be accepted in resource updated notifications.""" |
84 | | - params = types.ResourceUpdatedNotificationParams(uri="users/me") |
85 | | - assert str(params.uri) == "users/me" |
| 15 | +from mcp import types |
| 16 | +from mcp.server.lowlevel import Server |
| 17 | +from mcp.server.lowlevel.helper_types import ReadResourceContents |
| 18 | +from mcp.shared.memory import ( |
| 19 | + create_connected_server_and_client_session as client_session, |
| 20 | +) |
| 21 | + |
| 22 | +pytestmark = pytest.mark.anyio |
| 23 | + |
| 24 | + |
| 25 | +async def test_relative_uri_roundtrip(): |
| 26 | + """Relative URIs survive the full server-client JSON-RPC roundtrip. |
| 27 | +
|
| 28 | + This is the critical regression test - if someone reintroduces AnyUrl, |
| 29 | + the server would fail to serialize resources with relative URIs, |
| 30 | + or the URI would be transformed during the roundtrip. |
| 31 | + """ |
| 32 | + server = Server("test") |
| 33 | + |
| 34 | + @server.list_resources() |
| 35 | + async def list_resources(): |
| 36 | + return [ |
| 37 | + types.Resource(name="user", uri="users/me"), |
| 38 | + types.Resource(name="config", uri="./config"), |
| 39 | + types.Resource(name="parent", uri="../parent/resource"), |
| 40 | + ] |
| 41 | + |
| 42 | + @server.read_resource() |
| 43 | + async def read_resource(uri: str): |
| 44 | + return [ |
| 45 | + ReadResourceContents( |
| 46 | + content=f"data for {uri}", |
| 47 | + mime_type="text/plain", |
| 48 | + ) |
| 49 | + ] |
| 50 | + |
| 51 | + async with client_session(server) as client: |
| 52 | + # List should return the exact URIs we specified |
| 53 | + resources = await client.list_resources() |
| 54 | + uri_map = {r.uri: r for r in resources.resources} |
| 55 | + |
| 56 | + assert "users/me" in uri_map, f"Expected 'users/me' in {list(uri_map.keys())}" |
| 57 | + assert "./config" in uri_map, f"Expected './config' in {list(uri_map.keys())}" |
| 58 | + assert "../parent/resource" in uri_map, f"Expected '../parent/resource' in {list(uri_map.keys())}" |
| 59 | + |
| 60 | + # Read should work with each relative URI and preserve it in the response |
| 61 | + for uri_str in ["users/me", "./config", "../parent/resource"]: |
| 62 | + result = await client.read_resource(uri_str) |
| 63 | + assert len(result.contents) == 1 |
| 64 | + assert result.contents[0].uri == uri_str |
| 65 | + |
| 66 | + |
| 67 | +async def test_custom_scheme_uri_roundtrip(): |
| 68 | + """Custom scheme URIs work through the protocol. |
| 69 | +
|
| 70 | + Some MCP servers use custom schemes like "custom://resource". |
| 71 | + These should work end-to-end. |
| 72 | + """ |
| 73 | + server = Server("test") |
| 74 | + |
| 75 | + @server.list_resources() |
| 76 | + async def list_resources(): |
| 77 | + return [ |
| 78 | + types.Resource(name="custom", uri="custom://my-resource"), |
| 79 | + types.Resource(name="file", uri="file:///path/to/file"), |
| 80 | + ] |
| 81 | + |
| 82 | + @server.read_resource() |
| 83 | + async def read_resource(uri: str): |
| 84 | + return [ReadResourceContents(content="data", mime_type="text/plain")] |
| 85 | + |
| 86 | + async with client_session(server) as client: |
| 87 | + resources = await client.list_resources() |
| 88 | + uri_map = {r.uri: r for r in resources.resources} |
| 89 | + |
| 90 | + assert "custom://my-resource" in uri_map |
| 91 | + assert "file:///path/to/file" in uri_map |
| 92 | + |
| 93 | + # Read with custom scheme |
| 94 | + result = await client.read_resource("custom://my-resource") |
| 95 | + assert len(result.contents) == 1 |
| 96 | + |
| 97 | + |
| 98 | +def test_uri_json_roundtrip_preserves_value(): |
| 99 | + """URI is preserved exactly through JSON serialization. |
| 100 | +
|
| 101 | + This catches any Pydantic validation or normalization that would |
| 102 | + alter the URI during the JSON-RPC message flow. |
| 103 | + """ |
| 104 | + test_uris = [ |
| 105 | + "users/me", |
| 106 | + "custom://resource", |
| 107 | + "./relative", |
| 108 | + "../parent", |
| 109 | + "file:///absolute/path", |
| 110 | + "https://example.com/path", |
| 111 | + ] |
| 112 | + |
| 113 | + for uri_str in test_uris: |
| 114 | + resource = types.Resource(name="test", uri=uri_str) |
| 115 | + json_data = resource.model_dump(mode="json") |
| 116 | + restored = types.Resource.model_validate(json_data) |
| 117 | + assert restored.uri == uri_str, f"URI mutated: {uri_str} -> {restored.uri}" |
| 118 | + |
| 119 | + |
| 120 | +def test_resource_contents_uri_json_roundtrip(): |
| 121 | + """TextResourceContents URI is preserved through JSON serialization.""" |
| 122 | + test_uris = ["users/me", "./relative", "custom://resource"] |
| 123 | + |
| 124 | + for uri_str in test_uris: |
| 125 | + contents = types.TextResourceContents( |
| 126 | + uri=uri_str, |
| 127 | + text="data", |
| 128 | + mimeType="text/plain", |
| 129 | + ) |
| 130 | + json_data = contents.model_dump(mode="json") |
| 131 | + restored = types.TextResourceContents.model_validate(json_data) |
| 132 | + assert restored.uri == uri_str, f"URI mutated: {uri_str} -> {restored.uri}" |
0 commit comments