modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: change Resource URI fields from AnyUrl to str (#1863)

Browse source
This commit is contained in:
Felix Weinberger 2026-01-16 08:58:57 +00:00 • committed by GitHub
commit cfb2909631
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 343 additions and 179 deletions
Download patch file Download diff file Expand all files Collapse all files

6
tests/issues/test_152_resource_mime_type.py
View file

@ -70,9 +70,9 @@ async def test_lowlevel_resource_mime_type():
# Create test resources with specific mime types
test_resources = [
types.Resource(uri=AnyUrl("test://image"), name="test image", mimeType="image/png"),
types.Resource(uri="test://image", name="test image", mimeType="image/png"),
types.Resource(
uri=AnyUrl("test://image_bytes"),
uri="test://image_bytes",
name="test image bytes",
mimeType="image/png",
),
@ -83,7 +83,7 @@ async def test_lowlevel_resource_mime_type():
return test_resources
@server.read_resource()
async def handle_read_resource(uri: AnyUrl):
async def handle_read_resource(uri: str):
if str(uri) == "test://image":
return [ReadResourceContents(content=base64_string, mime_type="image/png")]
elif str(uri) == "test://image_bytes":

132
tests/issues/test_1574_resource_uri_validation.py Normal file
View file

@ -0,0 +1,132 @@
"""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 types
from mcp.server.lowlevel import Server
from mcp.server.lowlevel.helper_types import ReadResourceContents
from mcp.shared.memory import (
create_connected_server_and_client_session as client_session,
)
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.
"""
server = Server("test")
@server.list_resources()
async def list_resources():
return [
types.Resource(name="user", uri="users/me"),
types.Resource(name="config", uri="./config"),
types.Resource(name="parent", uri="../parent/resource"),
]
@server.read_resource()
async def read_resource(uri: str):
return [
ReadResourceContents(
content=f"data for {uri}",
mime_type="text/plain",
)
]
async with client_session(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.
"""
server = Server("test")
@server.list_resources()
async def list_resources():
return [
types.Resource(name="custom", uri="custom://my-resource"),
types.Resource(name="file", uri="file:///path/to/file"),
]
@server.read_resource()
async def read_resource(uri: str):
return [ReadResourceContents(content="data", mime_type="text/plain")]
async with client_session(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",
mimeType="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}"

5
tests/issues/test_342_base64_encoding.py
View file

@ -13,7 +13,6 @@ import base64
from typing import cast
import pytest
from pydantic import AnyUrl
from mcp.server.lowlevel.helper_types import ReadResourceContents
from mcp.server.lowlevel.server import Server
@ -46,7 +45,7 @@ async def test_server_base64_encoding_issue():
# Register a resource handler that returns our test data
@server.read_resource()
async def read_resource(uri: AnyUrl) -> list[ReadResourceContents]:
async def read_resource(uri: str) -> list[ReadResourceContents]:
return [ReadResourceContents(content=binary_data, mime_type="application/octet-stream")]
# Get the handler directly from the server
@ -54,7 +53,7 @@ async def test_server_base64_encoding_issue():
# Create a request
request = ReadResourceRequest(
params=ReadResourceRequestParams(uri=AnyUrl("test://resource")),
params=ReadResourceRequestParams(uri="test://resource"),
)
# Call the handler to get the response

13
tests/server/fastmcp/prompts/test_base.py
View file

@ -1,7 +1,6 @@
from typing import Any
import pytest
from pydantic import FileUrl
from mcp.server.fastmcp.prompts.base import AssistantMessage, Message, Prompt, TextContent, UserMessage
from mcp.types import EmbeddedResource, TextResourceContents
@ -95,7 +94,7 @@ class TestRenderPrompt:
content=EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=FileUrl("file://file.txt"),
uri="file://file.txt",
text="File contents",
mimeType="text/plain",
),
@ -108,7 +107,7 @@ class TestRenderPrompt:
content=EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=FileUrl("file://file.txt"),
uri="file://file.txt",
text="File contents",
mimeType="text/plain",
),
@ -127,7 +126,7 @@ class TestRenderPrompt:
content=EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=FileUrl("file://file.txt"),
uri="file://file.txt",
text="File contents",
mimeType="text/plain",
),
@ -143,7 +142,7 @@ class TestRenderPrompt:
content=EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=FileUrl("file://file.txt"),
uri="file://file.txt",
text="File contents",
mimeType="text/plain",
),
@ -162,7 +161,7 @@ class TestRenderPrompt:
"content": {
"type": "resource",
"resource": {
"uri": FileUrl("file://file.txt"),
"uri": "file://file.txt",
"text": "File contents",
"mimeType": "text/plain",
},
@ -175,7 +174,7 @@ class TestRenderPrompt:
content=EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=FileUrl("file://file.txt"),
uri="file://file.txt",
text="File contents",
mimeType="text/plain",
),

15
tests/server/fastmcp/resources/test_file_resources.py
View file

@ -3,7 +3,6 @@ from pathlib import Path
from tempfile import NamedTemporaryFile
import pytest
from pydantic import FileUrl
from mcp.server.fastmcp.resources import FileResource
@ -31,7 +30,7 @@ class TestFileResource:
def test_file_resource_creation(self, temp_file: Path):
"""Test creating a FileResource."""
resource = FileResource(
uri=FileUrl(temp_file.as_uri()),
uri=temp_file.as_uri(),
name="test",
description="test file",
path=temp_file,
@ -46,7 +45,7 @@ class TestFileResource:
def test_file_resource_str_path_conversion(self, temp_file: Path):
"""Test FileResource handles string paths."""
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
uri=f"file://{temp_file}",
name="test",
path=Path(str(temp_file)),
)
@ -57,7 +56,7 @@ class TestFileResource:
async def test_read_text_file(self, temp_file: Path):
"""Test reading a text file."""
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
uri=f"file://{temp_file}",
name="test",
path=temp_file,
)
@ -69,7 +68,7 @@ class TestFileResource:
async def test_read_binary_file(self, temp_file: Path):
"""Test reading a file as binary."""
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
uri=f"file://{temp_file}",
name="test",
path=temp_file,
is_binary=True,
@ -82,7 +81,7 @@ class TestFileResource:
"""Test error on relative path."""
with pytest.raises(ValueError, match="Path must be absolute"):
FileResource(
uri=FileUrl("file:///test.txt"),
uri="file:///test.txt",
name="test",
path=Path("test.txt"),
)
@ -93,7 +92,7 @@ class TestFileResource:
# Create path to non-existent file
missing = temp_file.parent / "missing.txt"
resource = FileResource(
uri=FileUrl("file:///missing.txt"),
uri="file:///missing.txt",
name="test",
path=missing,
)
@ -107,7 +106,7 @@ class TestFileResource:
temp_file.chmod(0o000) # Remove all permissions
try:
resource = FileResource(
uri=FileUrl(temp_file.as_uri()),
uri=temp_file.as_uri(),
name="test",
path=temp_file,
)

20
tests/server/fastmcp/resources/test_function_resources.py
View file

@ -1,5 +1,5 @@
import pytest
from pydantic import AnyUrl, BaseModel
from pydantic import BaseModel
from mcp.server.fastmcp.resources import FunctionResource
@ -14,7 +14,7 @@ class TestFunctionResource:
return "test content"
resource = FunctionResource(
uri=AnyUrl("fn://test"),
uri="fn://test",
name="test",
description="test function",
fn=my_func,
@ -33,7 +33,7 @@ class TestFunctionResource:
return "Hello, world!"
resource = FunctionResource(
uri=AnyUrl("function://test"),
uri="function://test",
name="test",
fn=get_data,
)
@ -49,7 +49,7 @@ class TestFunctionResource:
return b"Hello, world!"
resource = FunctionResource(
uri=AnyUrl("function://test"),
uri="function://test",
name="test",
fn=get_data,
)
@ -64,7 +64,7 @@ class TestFunctionResource:
return {"key": "value"}
resource = FunctionResource(
uri=AnyUrl("function://test"),
uri="function://test",
name="test",
fn=get_data,
)
@ -80,7 +80,7 @@ class TestFunctionResource:
raise ValueError("Test error")
resource = FunctionResource(
uri=AnyUrl("function://test"),
uri="function://test",
name="test",
fn=failing_func,
)
@ -95,7 +95,7 @@ class TestFunctionResource:
name: str
resource = FunctionResource(
uri=AnyUrl("function://test"),
uri="function://test",
name="test",
fn=lambda: MyModel(name="test"),
)
@ -114,7 +114,7 @@ class TestFunctionResource:
return CustomData()
resource = FunctionResource(
uri=AnyUrl("function://test"),
uri="function://test",
name="test",
fn=get_data,
)
@ -129,7 +129,7 @@ class TestFunctionResource:
return "Hello, world!"
resource = FunctionResource(
uri=AnyUrl("function://test"),
uri="function://test",
name="test",
fn=get_data,
)
@ -154,7 +154,7 @@ class TestFunctionResource:
assert resource.description == "get_data returns a string"
assert resource.mime_type == "text/plain"
assert resource.name == "test"
assert resource.uri == AnyUrl("function://test")
assert resource.uri == "function://test"
class TestFunctionResourceMetadata:

16
tests/server/fastmcp/resources/test_resource_manager.py
View file

@ -2,7 +2,7 @@ from pathlib import Path
from tempfile import NamedTemporaryFile
import pytest
from pydantic import AnyUrl, FileUrl
from pydantic import AnyUrl
from mcp.server.fastmcp.resources import FileResource, FunctionResource, ResourceManager, ResourceTemplate
@ -31,7 +31,7 @@ class TestResourceManager:
"""Test adding a resource."""
manager = ResourceManager()
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
uri=f"file://{temp_file}",
name="test",
path=temp_file,
)
@ -43,7 +43,7 @@ class TestResourceManager:
"""Test adding the same resource twice."""
manager = ResourceManager()
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
uri=f"file://{temp_file}",
name="test",
path=temp_file,
)
@ -56,7 +56,7 @@ class TestResourceManager:
"""Test warning on duplicate resources."""
manager = ResourceManager()
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
uri=f"file://{temp_file}",
name="test",
path=temp_file,
)
@ -68,7 +68,7 @@ class TestResourceManager:
"""Test disabling warning on duplicate resources."""
manager = ResourceManager(warn_on_duplicate_resources=False)
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
uri=f"file://{temp_file}",
name="test",
path=temp_file,
)
@ -81,7 +81,7 @@ class TestResourceManager:
"""Test getting a resource by URI."""
manager = ResourceManager()
resource = FileResource(
uri=FileUrl(f"file://{temp_file}"),
uri=f"file://{temp_file}",
name="test",
path=temp_file,
)
@ -120,12 +120,12 @@ class TestResourceManager:
"""Test listing all resources."""
manager = ResourceManager()
resource1 = FileResource(
uri=FileUrl(f"file://{temp_file}"),
uri=f"file://{temp_file}",
name="test1",
path=temp_file,
)
resource2 = FileResource(
uri=FileUrl(f"file://{temp_file}2"),
uri=f"file://{temp_file}2",
name="test2",
path=temp_file,
)

51
tests/server/fastmcp/resources/test_resources.py
View file

@ -1,5 +1,4 @@
import pytest
from pydantic import AnyUrl
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.resources import FunctionResource, Resource
@ -9,35 +8,35 @@ from mcp.types import Annotations
class TestResourceValidation:
"""Test base Resource validation."""
def test_resource_uri_validation(self):
"""Test URI validation."""
def test_resource_uri_accepts_any_string(self):
"""Test that URI field accepts any string per MCP spec."""
def dummy_func() -> str: # pragma: no cover
return "data"
# Valid URI
resource = FunctionResource(
uri=AnyUrl("http://example.com/data"),
uri="http://example.com/data",
name="test",
fn=dummy_func,
)
assert str(resource.uri) == "http://example.com/data"
assert resource.uri == "http://example.com/data"
# Missing protocol
with pytest.raises(ValueError, match="Input should be a valid URL"):
FunctionResource(
uri=AnyUrl("invalid"),
name="test",
fn=dummy_func,
)
# Relative path - now accepted per MCP spec
resource = FunctionResource(
uri="users/me",
name="test",
fn=dummy_func,
)
assert resource.uri == "users/me"
# Missing host
with pytest.raises(ValueError, match="Input should be a valid URL"):
FunctionResource(
uri=AnyUrl("http://"),
name="test",
fn=dummy_func,
)
# Custom scheme
resource = FunctionResource(
uri="custom://resource",
name="test",
fn=dummy_func,
)
assert resource.uri == "custom://resource"
def test_resource_name_from_uri(self):
"""Test name is extracted from URI if not provided."""
@ -46,7 +45,7 @@ class TestResourceValidation:
return "data"
resource = FunctionResource(
uri=AnyUrl("resource://my-resource"),
uri="resource://my-resource",
fn=dummy_func,
)
assert resource.name == "resource://my-resource"
@ -65,7 +64,7 @@ class TestResourceValidation:
# Explicit name takes precedence over URI
resource = FunctionResource(
uri=AnyUrl("resource://uri-name"),
uri="resource://uri-name",
name="explicit-name",
fn=dummy_func,
)
@ -79,14 +78,14 @@ class TestResourceValidation:
# Default mime type
resource = FunctionResource(
uri=AnyUrl("resource://test"),
uri="resource://test",
fn=dummy_func,
)
assert resource.mime_type == "text/plain"
# Custom mime type
resource = FunctionResource(
uri=AnyUrl("resource://test"),
uri="resource://test",
fn=dummy_func,
mime_type="application/json",
)
@ -100,7 +99,7 @@ class TestResourceValidation:
pass
with pytest.raises(TypeError, match="abstract method"):
ConcreteResource(uri=AnyUrl("test://test"), name="test") # type: ignore
ConcreteResource(uri="test://test", name="test") # type: ignore
class TestResourceAnnotations:
@ -207,7 +206,7 @@ class TestResourceMetadata:
metadata = {"version": "1.0", "category": "test"}
resource = FunctionResource(
uri=AnyUrl("resource://test"),
uri="resource://test",
name="test",
fn=dummy_func,
meta=metadata,
@ -225,7 +224,7 @@ class TestResourceMetadata:
return "data"
resource = FunctionResource(
uri=AnyUrl("resource://test"),
uri="resource://test",
name="test",
fn=dummy_func,
)

38
tests/server/fastmcp/test_server.py
View file

@ -4,7 +4,7 @@ from typing import Any
from unittest.mock import patch
import pytest
from pydantic import AnyUrl, BaseModel
from pydantic import BaseModel
from starlette.routing import Mount, Route
from mcp.server.fastmcp import Context, FastMCP
@ -743,11 +743,11 @@ class TestServerResources:
def get_text():
return "Hello, world!"
resource = FunctionResource(uri=AnyUrl("resource://test"), name="test", fn=get_text)
resource = FunctionResource(uri="resource://test", name="test", fn=get_text)
mcp.add_resource(resource)
async with client_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://test"))
result = await client.read_resource("resource://test")
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[0].text == "Hello, world!"
@ -759,7 +759,7 @@ class TestServerResources:
return b"Binary data"
resource = FunctionResource(
uri=AnyUrl("resource://binary"),
uri="resource://binary",
name="binary",
fn=get_binary,
mime_type="application/octet-stream",
@ -767,7 +767,7 @@ class TestServerResources:
mcp.add_resource(resource)
async with client_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://binary"))
result = await client.read_resource("resource://binary")
assert isinstance(result.contents[0], BlobResourceContents)
assert result.contents[0].blob == base64.b64encode(b"Binary data").decode()
@ -779,11 +779,11 @@ class TestServerResources:
text_file = tmp_path / "test.txt"
text_file.write_text("Hello from file!")
resource = FileResource(uri=AnyUrl("file://test.txt"), name="test.txt", path=text_file)
resource = FileResource(uri="file://test.txt", name="test.txt", path=text_file)
mcp.add_resource(resource)
async with client_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("file://test.txt"))
result = await client.read_resource("file://test.txt")
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[0].text == "Hello from file!"
@ -796,7 +796,7 @@ class TestServerResources:
binary_file.write_bytes(b"Binary file data")
resource = FileResource(
uri=AnyUrl("file://test.bin"),
uri="file://test.bin",
name="test.bin",
path=binary_file,
mime_type="application/octet-stream",
@ -804,7 +804,7 @@ class TestServerResources:
mcp.add_resource(resource)
async with client_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("file://test.bin"))
result = await client.read_resource("file://test.bin")
assert isinstance(result.contents[0], BlobResourceContents)
assert result.contents[0].blob == base64.b64encode(b"Binary file data").decode()
@ -822,7 +822,7 @@ class TestServerResources:
assert len(resources.resources) == 1
resource = resources.resources[0]
assert resource.description == "get_data returns a string"
assert resource.uri == AnyUrl("function://test")
assert resource.uri == "function://test"
assert resource.name == "test_get_data"
assert resource.mimeType == "text/plain"
@ -870,7 +870,7 @@ class TestServerResourceTemplates:
return f"Data for {name}"
async with client_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://test/data"))
result = await client.read_resource("resource://test/data")
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[0].text == "Data for test"
@ -895,7 +895,7 @@ class TestServerResourceTemplates:
return f"Data for {org}/{repo}"
async with client_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://cursor/fastmcp/data"))
result = await client.read_resource("resource://cursor/fastmcp/data")
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[0].text == "Data for cursor/fastmcp"
@ -918,7 +918,7 @@ class TestServerResourceTemplates:
return "Static data"
async with client_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://static"))
result = await client.read_resource("resource://static")
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[0].text == "Static data"
@ -958,7 +958,7 @@ class TestServerResourceTemplates:
assert template.mimeType == "text/csv"
async with client_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://bob/csv"))
result = await client.read_resource("resource://bob/csv")
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[0].text == "csv for bob"
@ -1020,7 +1020,7 @@ class TestServerResourceMetadata:
return "test data"
async with client_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://data"))
result = await client.read_resource("resource://data")
# Verify content and metadata in protocol response
assert isinstance(result.contents[0], TextResourceContents)
@ -1189,7 +1189,7 @@ class TestContextInjection:
# Test via client
async with client_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://context/test"))
result = await client.read_resource("resource://context/test")
assert len(result.contents) == 1
content = result.contents[0]
assert isinstance(content, TextResourceContents)
@ -1214,7 +1214,7 @@ class TestContextInjection:
# Test via client
async with client_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://nocontext/test"))
result = await client.read_resource("resource://nocontext/test")
assert len(result.contents) == 1
content = result.contents[0]
assert isinstance(content, TextResourceContents)
@ -1239,7 +1239,7 @@ class TestContextInjection:
# Test via client
async with client_session(mcp._mcp_server) as client:
result = await client.read_resource(AnyUrl("resource://custom/123"))
result = await client.read_resource("resource://custom/123")
assert len(result.contents) == 1
content = result.contents[0]
assert isinstance(content, TextResourceContents)
@ -1442,7 +1442,7 @@ class TestServerPrompts:
content=EmbeddedResource(
type="resource",
resource=TextResourceContents(
uri=AnyUrl("file://file.txt"),
uri="file://file.txt",
text="File contents",
mimeType="text/plain",
),

9
tests/server/fastmcp/test_title.py
View file

@ -1,7 +1,6 @@
"""Integration tests for title field functionality."""
import pytest
from pydantic import AnyUrl
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.resources import FunctionResource
@ -134,7 +133,7 @@ async def test_resource_title():
return "Basic data"
basic_resource = FunctionResource(
uri=AnyUrl("resource://basic"),
uri="resource://basic",
name="basic_resource",
description="Basic resource",
fn=get_basic_data,
@ -146,7 +145,7 @@ async def test_resource_title():
return "Titled data"
titled_resource = FunctionResource(
uri=AnyUrl("resource://titled"),
uri="resource://titled",
name="titled_resource",
title="User-Friendly Resource",
description="Resource with title",
@ -219,10 +218,10 @@ async def test_get_display_name_utility():
assert get_display_name(tool_with_both) == "Primary Title"
# Test other types: title > name
resource = Resource(uri=AnyUrl("file://test"), name="test_res")
resource = Resource(uri="file://test", name="test_res")
assert get_display_name(resource) == "test_res"
resource_with_title = Resource(uri=AnyUrl("file://test"), name="test_res", title="Test Resource")
resource_with_title = Resource(uri="file://test", name="test_res", title="Test Resource")
assert get_display_name(resource_with_title) == "Test Resource"
prompt = Prompt(name="test_prompt")

5
tests/server/lowlevel/test_server_listing.py
View file

@ -3,7 +3,6 @@
import warnings
import pytest
from pydantic import AnyUrl
from mcp.server import Server
from mcp.types import (
@ -52,8 +51,8 @@ async def test_list_resources_basic() -> None:
server = Server("test")
test_resources = [
Resource(uri=AnyUrl("file:///test1.txt"), name="Test 1"),
Resource(uri=AnyUrl("file:///test2.txt"), name="Test 2"),
Resource(uri="file:///test1.txt", name="Test 1"),
Resource(uri="file:///test2.txt", name="Test 2"),
]
with warnings.catch_warnings():

13
tests/server/test_read_resource.py
View file

@ -3,7 +3,6 @@ from pathlib import Path
from tempfile import NamedTemporaryFile
import pytest
from pydantic import AnyUrl, FileUrl
import mcp.types as types
from mcp.server.lowlevel.server import ReadResourceContents, Server
@ -27,7 +26,7 @@ async def test_read_resource_text(temp_file: Path):
server = Server("test")
@server.read_resource()
async def read_resource(uri: AnyUrl) -> Iterable[ReadResourceContents]:
async def read_resource(uri: str) -> Iterable[ReadResourceContents]:
return [ReadResourceContents(content="Hello World", mime_type="text/plain")]
# Get the handler directly from the server
@ -35,7 +34,7 @@ async def test_read_resource_text(temp_file: Path):
# Create a request
request = types.ReadResourceRequest(
params=types.ReadResourceRequestParams(uri=FileUrl(temp_file.as_uri())),
params=types.ReadResourceRequestParams(uri=temp_file.as_uri()),
)
# Call the handler
@ -54,7 +53,7 @@ async def test_read_resource_binary(temp_file: Path):
server = Server("test")
@server.read_resource()
async def read_resource(uri: AnyUrl) -> Iterable[ReadResourceContents]:
async def read_resource(uri: str) -> Iterable[ReadResourceContents]:
return [ReadResourceContents(content=b"Hello World", mime_type="application/octet-stream")]
# Get the handler directly from the server
@ -62,7 +61,7 @@ async def test_read_resource_binary(temp_file: Path):
# Create a request
request = types.ReadResourceRequest(
params=types.ReadResourceRequestParams(uri=FileUrl(temp_file.as_uri())),
params=types.ReadResourceRequestParams(uri=temp_file.as_uri()),
)
# Call the handler
@ -80,7 +79,7 @@ async def test_read_resource_default_mime(temp_file: Path):
server = Server("test")
@server.read_resource()
async def read_resource(uri: AnyUrl) -> Iterable[ReadResourceContents]:
async def read_resource(uri: str) -> Iterable[ReadResourceContents]:
return [
ReadResourceContents(
content="Hello World",
@ -93,7 +92,7 @@ async def test_read_resource_default_mime(temp_file: Path):
# Create a request
request = types.ReadResourceRequest(
params=types.ReadResourceRequestParams(uri=FileUrl(temp_file.as_uri())),
params=types.ReadResourceRequestParams(uri=temp_file.as_uri()),
)
# Call the handler

3
tests/shared/test_memory.py
View file

@ -1,5 +1,4 @@
import pytest
from pydantic import AnyUrl
from typing_extensions import AsyncGenerator
from mcp.client.session import ClientSession
@ -16,7 +15,7 @@ def mcp_server() -> Server:
async def handle_list_resources(): # pragma: no cover
return [
Resource(
uri=AnyUrl("memory://test"),
uri="memory://test",
name="Test Resource",
description="A test resource",
)

21
tests/shared/test_sse.py
View file

@ -5,6 +5,7 @@ import time
from collections.abc import AsyncGenerator, Generator
from typing import Any
from unittest.mock import AsyncMock, MagicMock, Mock, patch
from urllib.parse import urlparse
import anyio
import httpx
@ -12,7 +13,6 @@ import pytest
import uvicorn
from httpx_sse import ServerSentEvent
from inline_snapshot import snapshot
from pydantic import AnyUrl
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import Response
@ -61,13 +61,14 @@ class ServerTest(Server): # pragma: no cover
super().__init__(SERVER_NAME)
@self.read_resource()
async def handle_read_resource(uri: AnyUrl) -> str | bytes:
if uri.scheme == "foobar":
return f"Read {uri.host}"
elif uri.scheme == "slow":
async def handle_read_resource(uri: str) -> str | bytes:
parsed = urlparse(uri)
if parsed.scheme == "foobar":
return f"Read {parsed.netloc}"
if parsed.scheme == "slow":
# Simulate a slow resource
await anyio.sleep(2.0)
return f"Slow response from {uri.host}"
return f"Slow response from {parsed.netloc}"
raise McpError(error=ErrorData(code=404, message="OOPS! no resource with that URI was found"))
@ -254,7 +255,7 @@ async def test_sse_client_happy_request_and_response(
initialized_sse_client_session: ClientSession,
) -> None:
session = initialized_sse_client_session
response = await session.read_resource(uri=AnyUrl("foobar://should-work"))
response = await session.read_resource(uri="foobar://should-work")
assert len(response.contents) == 1
assert isinstance(response.contents[0], TextResourceContents)
assert response.contents[0].text == "Read should-work"
@ -266,7 +267,7 @@ async def test_sse_client_exception_handling(
) -> None:
session = initialized_sse_client_session
with pytest.raises(McpError, match="OOPS! no resource with that URI was found"):
await session.read_resource(uri=AnyUrl("xxx://will-not-work"))
await session.read_resource(uri="xxx://will-not-work")
@pytest.mark.anyio
@ -277,12 +278,12 @@ async def test_sse_client_timeout( # pragma: no cover
session = initialized_sse_client_session
# sanity check that normal, fast responses are working
response = await session.read_resource(uri=AnyUrl("foobar://1"))
response = await session.read_resource(uri="foobar://1")
assert isinstance(response, ReadResourceResult)
with anyio.move_on_after(3):
with pytest.raises(McpError, match="Read timed out"):
response = await session.read_resource(uri=AnyUrl("slow://2"))
response = await session.read_resource(uri="slow://2")
# we should receive an error here
return

33
tests/shared/test_streamable_http.py
View file

@ -13,6 +13,7 @@ import time
from collections.abc import Generator
from typing import Any
from unittest.mock import MagicMock
from urllib.parse import urlparse
import anyio
import httpx
@ -20,7 +21,6 @@ import pytest
import requests
import uvicorn
from httpx_sse import ServerSentEvent
from pydantic import AnyUrl
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.routing import Mount
@ -137,13 +137,14 @@ class ServerTest(Server): # pragma: no cover
self._lock = None # Will be initialized in async context
@self.read_resource()
async def handle_read_resource(uri: AnyUrl) -> str | bytes:
if uri.scheme == "foobar":
return f"Read {uri.host}"
elif uri.scheme == "slow":
async def handle_read_resource(uri: str) -> str | bytes:
parsed = urlparse(uri)
if parsed.scheme == "foobar":
return f"Read {parsed.netloc}"
if parsed.scheme == "slow":
# Simulate a slow resource
await anyio.sleep(2.0)
return f"Slow response from {uri.host}"
return f"Slow response from {parsed.netloc}"
raise ValueError(f"Unknown resource: {uri}")
@ -214,7 +215,7 @@ class ServerTest(Server): # pragma: no cover
# When the tool is called, send a notification to test GET stream
if name == "test_tool_with_standalone_notification":
await ctx.session.send_resource_updated(uri=AnyUrl("http://test_resource"))
await ctx.session.send_resource_updated(uri="http://test_resource")
return [TextContent(type="text", text=f"Called {name}")]
elif name == "long_running_with_checkpoints":
@ -368,7 +369,7 @@ class ServerTest(Server): # pragma: no cover
elif name == "tool_with_standalone_stream_close":
# Test for GET stream reconnection
# 1. Send unsolicited notification via GET stream (no related_request_id)
await ctx.session.send_resource_updated(uri=AnyUrl("http://notification_1"))
await ctx.session.send_resource_updated(uri="http://notification_1")
# Small delay to ensure notification is flushed before closing
await anyio.sleep(0.1)
@ -381,7 +382,7 @@ class ServerTest(Server): # pragma: no cover
await anyio.sleep(1.5)
# 4. Send another notification on the new GET stream connection
await ctx.session.send_resource_updated(uri=AnyUrl("http://notification_2"))
await ctx.session.send_resource_updated(uri="http://notification_2")
return [TextContent(type="text", text="Standalone stream close test done")]
@ -1009,9 +1010,9 @@ async def test_streamable_http_client_basic_connection(basic_server: None, basic
@pytest.mark.anyio
async def test_streamable_http_client_resource_read(initialized_client_session: ClientSession):
"""Test client resource read functionality."""
response = await initialized_client_session.read_resource(uri=AnyUrl("foobar://test-resource"))
response = await initialized_client_session.read_resource(uri="foobar://test-resource")
assert len(response.contents) == 1
assert response.contents[0].uri == AnyUrl("foobar://test-resource")
assert response.contents[0].uri == "foobar://test-resource"
assert isinstance(response.contents[0], TextResourceContents)
assert response.contents[0].text == "Read test-resource"
@ -1035,7 +1036,7 @@ async def test_streamable_http_client_tool_invocation(initialized_client_session
async def test_streamable_http_client_error_handling(initialized_client_session: ClientSession):
"""Test error handling in client."""
with pytest.raises(McpError) as exc_info:
await initialized_client_session.read_resource(uri=AnyUrl("unknown://test-error"))
await initialized_client_session.read_resource(uri="unknown://test-error")
assert exc_info.value.error.code == 0
assert "Unknown resource: unknown://test-error" in exc_info.value.error.message
@ -1061,7 +1062,7 @@ async def test_streamable_http_client_session_persistence(basic_server: None, ba
assert len(tools.tools) == 10
# Read a resource
resource = await session.read_resource(uri=AnyUrl("foobar://test-persist"))
resource = await session.read_resource(uri="foobar://test-persist")
assert isinstance(resource.contents[0], TextResourceContents) is True
content = resource.contents[0]
assert isinstance(content, TextResourceContents)
@ -1130,7 +1131,7 @@ async def test_streamable_http_client_get_stream(basic_server: None, basic_serve
resource_update_found = False
for notif in notifications_received:
if isinstance(notif.root, types.ResourceUpdatedNotification): # pragma: no branch
assert str(notif.root.params.uri) == "http://test_resource/"
assert str(notif.root.params.uri) == "http://test_resource"
resource_update_found = True
assert resource_update_found, "ResourceUpdatedNotification not received via GET stream"
@ -2235,10 +2236,10 @@ async def test_standalone_get_stream_reconnection(
assert result.content[0].text == "Standalone stream close test done"
# Verify both notifications were received
assert "http://notification_1/" in received_notifications, (
assert "http://notification_1" in received_notifications, (
f"Should receive notification 1 (sent before GET stream close), got: {received_notifications}"
)
assert "http://notification_2/" in received_notifications, (
assert "http://notification_2" in received_notifications, (
f"Should receive notification 2 after reconnect, got: {received_notifications}"
)

13
tests/shared/test_ws.py
View file

@ -49,13 +49,16 @@ class ServerTest(Server): # pragma: no cover
super().__init__(SERVER_NAME)
@self.read_resource()
async def handle_read_resource(uri: AnyUrl) -> str | bytes:
if uri.scheme == "foobar":
return f"Read {uri.host}"
elif uri.scheme == "slow":
async def handle_read_resource(uri: str) -> str | bytes:
from urllib.parse import urlparse
parsed = urlparse(uri)
if parsed.scheme == "foobar":
return f"Read {parsed.netloc}"
elif parsed.scheme == "slow":
# Simulate a slow resource
await anyio.sleep(2.0)
return f"Slow response from {uri.host}"
return f"Slow response from {parsed.netloc}"
raise McpError(error=ErrorData(code=404, message="OOPS! no resource with that URI was found"))