-
Notifications
You must be signed in to change notification settings - Fork 3.3k
Expand file tree
/
Copy pathtest_1574_resource_uri_validation.py
More file actions
140 lines (109 loc) · 5.23 KB
/
test_1574_resource_uri_validation.py
File metadata and controls
140 lines (109 loc) · 5.23 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
"""Tests for issue #1574: Python SDK incorrectly validates Resource URIs.
The Python SDK previously used Pydantic's AnyUrl for URI fields, which rejected
relative paths like 'users/me' that are valid according to the MCP spec and
accepted by the TypeScript SDK.
The fix changed URI fields to plain strings to match the spec, which defines
uri fields as strings with no JSON Schema format validation.
These tests verify the fix works end-to-end through the JSON-RPC protocol.
"""
import pytest
from mcp import Client, types
from mcp.server import Server, ServerRequestContext
from mcp.types import (
ListResourcesResult,
PaginatedRequestParams,
ReadResourceRequestParams,
ReadResourceResult,
TextResourceContents,
)
pytestmark = pytest.mark.anyio
async def test_relative_uri_roundtrip():
"""Relative URIs survive the full server-client JSON-RPC roundtrip.
This is the critical regression test - if someone reintroduces AnyUrl,
the server would fail to serialize resources with relative URIs,
or the URI would be transformed during the roundtrip.
"""
async def handle_list_resources(
ctx: ServerRequestContext, params: PaginatedRequestParams | None
) -> ListResourcesResult:
return ListResourcesResult(
resources=[
types.Resource(name="user", uri="users/me"),
types.Resource(name="config", uri="./config"),
types.Resource(name="parent", uri="../parent/resource"),
]
)
async def handle_read_resource(ctx: ServerRequestContext, params: ReadResourceRequestParams) -> ReadResourceResult:
return ReadResourceResult(
contents=[TextResourceContents(uri=str(params.uri), text=f"data for {params.uri}", mime_type="text/plain")]
)
server = Server("test", on_list_resources=handle_list_resources, on_read_resource=handle_read_resource)
async with Client(server) as client:
# List should return the exact URIs we specified
resources = await client.list_resources()
uri_map = {r.uri: r for r in resources.resources}
assert "users/me" in uri_map, f"Expected 'users/me' in {list(uri_map.keys())}"
assert "./config" in uri_map, f"Expected './config' in {list(uri_map.keys())}"
assert "../parent/resource" in uri_map, f"Expected '../parent/resource' in {list(uri_map.keys())}"
# Read should work with each relative URI and preserve it in the response
for uri_str in ["users/me", "./config", "../parent/resource"]:
result = await client.read_resource(uri_str)
assert len(result.contents) == 1
assert result.contents[0].uri == uri_str
async def test_custom_scheme_uri_roundtrip():
"""Custom scheme URIs work through the protocol.
Some MCP servers use custom schemes like "custom://resource".
These should work end-to-end.
"""
async def handle_list_resources(
ctx: ServerRequestContext, params: PaginatedRequestParams | None
) -> ListResourcesResult:
return ListResourcesResult(
resources=[
types.Resource(name="custom", uri="custom://my-resource"),
types.Resource(name="file", uri="file:///path/to/file"),
]
)
async def handle_read_resource(ctx: ServerRequestContext, params: ReadResourceRequestParams) -> ReadResourceResult:
return ReadResourceResult(
contents=[TextResourceContents(uri=str(params.uri), text="data", mime_type="text/plain")]
)
server = Server("test", on_list_resources=handle_list_resources, on_read_resource=handle_read_resource)
async with Client(server) as client:
resources = await client.list_resources()
uri_map = {r.uri: r for r in resources.resources}
assert "custom://my-resource" in uri_map
assert "file:///path/to/file" in uri_map
# Read with custom scheme
result = await client.read_resource("custom://my-resource")
assert len(result.contents) == 1
def test_uri_json_roundtrip_preserves_value():
"""URI is preserved exactly through JSON serialization.
This catches any Pydantic validation or normalization that would
alter the URI during the JSON-RPC message flow.
"""
test_uris = [
"users/me",
"custom://resource",
"./relative",
"../parent",
"file:///absolute/path",
"https://example.com/path",
]
for uri_str in test_uris:
resource = types.Resource(name="test", uri=uri_str)
json_data = resource.model_dump(mode="json")
restored = types.Resource.model_validate(json_data)
assert restored.uri == uri_str, f"URI mutated: {uri_str} -> {restored.uri}"
def test_resource_contents_uri_json_roundtrip():
"""TextResourceContents URI is preserved through JSON serialization."""
test_uris = ["users/me", "./relative", "custom://resource"]
for uri_str in test_uris:
contents = types.TextResourceContents(
uri=uri_str,
text="data",
mime_type="text/plain",
)
json_data = contents.model_dump(mode="json")
restored = types.TextResourceContents.model_validate(json_data)
assert restored.uri == uri_str, f"URI mutated: {uri_str} -> {restored.uri}"