Fix matching logic

This commit is contained in:
Jeremiah Lowin 2025-04-15 11:06:51 -04:00
commit c2ca6e1fa2
4 changed files with 303 additions and 19 deletions

View file

@ -203,8 +203,12 @@ class ResourceManager:
self._templates[storage_key] = template
return template
async def get_resource(self, uri: AnyUrl | str) -> Resource | None:
"""Get resource by URI, checking concrete resources first, then templates."""
async def get_resource(self, uri: AnyUrl | str) -> Resource:
"""Get resource by URI, checking concrete resources first, then templates.
Raises:
ResourceError: If no resource or template matching the URI is found.
"""
uri_str = str(uri)
logger.debug("Getting resource", extra={"uri": uri_str})
@ -221,7 +225,7 @@ class ResourceManager:
except Exception as e:
raise ValueError(f"Error creating resource from template: {e}")
raise ResourceError(f"Unknown resource: {uri}")
raise ResourceError(f"Unknown resource: {uri_str}")
def get_resources(self) -> dict[str, Resource]:
"""Get all registered resources, keyed by URI."""

View file

@ -22,22 +22,23 @@ from fastmcp.resources.types import FunctionResource, Resource
from fastmcp.utilities.types import _convert_set_defaults
def match_uri_template(uri: str, uri_template: str) -> dict[str, Any] | None:
"""Match a URI against a template and extract parameters.
def build_regex(template: str) -> re.Pattern:
# Escape all non-brace characters, then restore {var} placeholders
parts = re.split(r"(\{[^}]+\})", template)
pattern = ""
for part in parts:
if part.startswith("{") and part.endswith("}"):
name = part[1:-1]
pattern += f"(?P<{name}>[^/]+)"
else:
pattern += re.escape(part)
return re.compile(f"^{pattern}$")
Args:
uri: The URI to match against the template
uri_template: The URI template to match against
Returns:
A dictionary of extracted parameters if there's a match, or None if no match
"""
# Convert template to regex pattern
pattern = uri_template.replace("{", "(?P<").replace("}", ">[^/]+)")
match = re.match(f"^{pattern}$", uri)
if match:
return match.groupdict()
return None
def match_uri_template(uri: str, uri_template: str) -> dict[str, str] | None:
regex = build_regex(uri_template)
match = regex.match(uri)
return match.groupdict() if match else None
class MyModel(BaseModel):

View file

@ -43,6 +43,60 @@ def fastmcp_server():
return server
@pytest.fixture
def tagged_resources_server():
"""Fixture that creates a FastMCP server with tagged resources and templates."""
server = FastMCP("TaggedResourcesServer")
# Add a resource with tags
@server.resource(
uri="data://tagged", tags={"test", "metadata"}, description="A tagged resource"
)
async def get_tagged_data():
return {"type": "tagged_data"}
# Add a resource template with tags
@server.resource(
uri="template://{id}",
tags={"template", "parameterized"},
description="A tagged template",
)
async def get_template_data(id: str):
return {"id": id, "type": "template_data"}
return server
@pytest.fixture
def mounted_resources_server():
"""Fixture that creates a FastMCP server with mounted resources."""
# Create the main server
main_server = FastMCP("MainServer")
# Create sub-app with its own resources
sub_app = FastMCP("SubAppServer")
# Add a resource to the sub-app
@sub_app.resource(uri="subapp://data", description="SubApp resource")
async def get_subapp_data():
return {"source": "subapp"}
# Add a template to the sub-app
@sub_app.resource(uri="subapp://{id}", description="SubApp template")
async def get_subapp_item(id: str):
return {"id": id, "source": "subapp"}
# Mount the sub-app to the main server with a prefix
main_server.mount("sub", sub_app)
# Add a resource to the main server
@main_server.resource(uri="main://data", description="Main resource")
async def get_main_data():
return {"source": "main"}
return main_server
async def test_list_tools(fastmcp_server):
"""Test listing tools with InMemoryClient."""
client = Client(transport=FastMCPTransport(fastmcp_server))
@ -157,3 +211,146 @@ async def test_resource_template(fastmcp_server):
assert '"id": "123"' in content_str
assert '"name": "User 123"' in content_str
assert '"active": true' in content_str
async def test_mcp_resource_generation(fastmcp_server):
"""Test that resources are properly generated in MCP format."""
client = Client(transport=FastMCPTransport(fastmcp_server))
async with client:
resources = await client.list_resources()
assert len(resources) == 1
resource = resources[0]
# Verify resource has correct MCP format
assert hasattr(resource, "uri")
assert hasattr(resource, "name")
assert hasattr(resource, "description")
assert str(resource.uri) == "data://users"
async def test_mcp_template_generation(fastmcp_server):
"""Test that templates are properly generated in MCP format."""
client = Client(transport=FastMCPTransport(fastmcp_server))
async with client:
templates = await client.list_resource_templates()
assert len(templates) == 1
template = templates[0]
# Verify template has correct MCP format
assert hasattr(template, "uriTemplate")
assert hasattr(template, "name")
assert hasattr(template, "description")
assert "data://user/{user_id}" in template.uriTemplate
async def test_template_access_via_client(fastmcp_server):
"""Test that templates can be accessed through a client."""
client = Client(transport=FastMCPTransport(fastmcp_server))
async with client:
# Verify template works correctly when accessed
uri = cast(AnyUrl, "data://user/456")
result = await client.read_resource(uri)
content_str = str(result[0])
assert '"id": "456"' in content_str
async def test_tagged_resource_metadata(tagged_resources_server):
"""Test that resource metadata is preserved in MCP format."""
client = Client(transport=FastMCPTransport(tagged_resources_server))
async with client:
resources = await client.list_resources()
assert len(resources) == 1
resource = resources[0]
# Verify resource metadata is preserved
assert str(resource.uri) == "data://tagged"
assert resource.description == "A tagged resource"
async def test_tagged_template_metadata(tagged_resources_server):
"""Test that template metadata is preserved in MCP format."""
client = Client(transport=FastMCPTransport(tagged_resources_server))
async with client:
templates = await client.list_resource_templates()
assert len(templates) == 1
template = templates[0]
# Verify template metadata is preserved
assert "template://{id}" in template.uriTemplate
assert template.description == "A tagged template"
async def test_tagged_template_functionality(tagged_resources_server):
"""Test that tagged templates function correctly when accessed."""
client = Client(transport=FastMCPTransport(tagged_resources_server))
async with client:
# Verify template functionality
uri = cast(AnyUrl, "template://123")
result = await client.read_resource(uri)
content_str = str(result[0])
assert '"id": "123"' in content_str
assert '"type": "template_data"' in content_str
async def test_mounted_resources(mounted_resources_server):
"""Test that resources from mounted apps are correctly prefixed."""
client = Client(transport=FastMCPTransport(mounted_resources_server))
async with client:
resources = await client.list_resources()
# Should have two resources (one from main, one from sub)
assert len(resources) == 2
# Find resources by URI
main_resource = next(
(r for r in resources if str(r.uri) == "main://data"), None
)
sub_resource = next(
(r for r in resources if str(r.uri) == "sub+subapp://data"), None
)
# Both resources should exist
assert main_resource is not None
assert sub_resource is not None
# Check descriptions
assert main_resource.description == "Main resource"
assert sub_resource.description == "SubApp resource"
async def test_mounted_templates(mounted_resources_server):
"""Test that templates from mounted apps are correctly prefixed."""
client = Client(transport=FastMCPTransport(mounted_resources_server))
async with client:
templates = await client.list_resource_templates()
# Should have one template (from sub)
assert len(templates) == 1
# Check the template
template = templates[0]
assert "sub+subapp://{id}" in template.uriTemplate
assert template.description == "SubApp template"
async def test_mounted_template_functionality(mounted_resources_server):
"""Test that templates from mounted apps function correctly."""
client = Client(transport=FastMCPTransport(mounted_resources_server))
async with client:
# Use the prefixed template
uri = cast(AnyUrl, "sub+subapp://123")
result = await client.read_resource(uri)
content_str = str(result[0])
# Check the content
assert '"id": "123"' in content_str
assert '"source": "subapp"' in content_str

View file

@ -4,6 +4,7 @@ import pytest
from pydantic import BaseModel
from fastmcp.resources import FunctionResource, ResourceTemplate
from fastmcp.resources.template import match_uri_template
class TestResourceTemplate:
@ -46,6 +47,27 @@ class TestResourceTemplate:
assert template.matches("test://foo") is None
assert template.matches("other://foo/123") is None
def test_template_matches_with_prefix(self):
"""Test matching URIs against a template with a prefix."""
def my_func(key: str, value: int) -> dict:
return {"key": key, "value": value}
template = ResourceTemplate.from_function(
fn=my_func,
uri_template="app+test://{key}/{value}",
name="test",
)
# Valid match
params = template.matches("app+test://foo/123")
assert params == {"key": "foo", "value": "123"}
# No match
assert template.matches("test://foo/123") is None
assert template.matches("test://foo") is None
assert template.matches("other://foo/123") is None
def test_template_uri_validation(self):
"""Test validation rule: URI template must have at least one parameter."""
@ -279,3 +301,63 @@ class TestResourceTemplate:
assert isinstance(resource, FunctionResource)
content = await resource.read()
assert content == "hello"
class TestMatchUriTemplate:
"""Test match_uri_template function."""
@pytest.mark.parametrize(
"uri, expected_params",
[
("test://foo/123", {"x": "foo", "y": "123"}),
("test://bar/456", {"x": "bar", "y": "456"}),
("test://foo/bar", {"x": "foo", "y": "bar"}),
("prefix+test://foo/123", None),
("test://foo", None),
("other://foo/123", None),
("t.est://foo/bar", None),
],
)
def test_match_uri_template_simple_params(
self, uri: str, expected_params: dict[str, str] | None
):
"""Test matching URIs against a template with simple parameters."""
uri_template = "test://{x}/{y}"
result = match_uri_template(uri=uri, uri_template=uri_template)
assert result == expected_params
@pytest.mark.parametrize(
"uri, expected_params",
[
("test://a/b/foo/c/d/123", {"x": "foo", "y": "123"}),
("test://a/b/bar/c/d/456", {"x": "bar", "y": "456"}),
("prefix+test://a/b/foo/c/d/123", None),
("test://a/b/foo", None),
("other://a/b/foo/c/d/123", None),
],
)
def test_match_uri_template_params_and_literal_segments(
self, uri: str, expected_params: dict[str, str] | None
):
"""Test matching URIs against a template with parameters and literal segments."""
uri_template = "test://a/b/{x}/c/d/{y}"
result = match_uri_template(uri=uri, uri_template=uri_template)
assert result == expected_params
@pytest.mark.parametrize(
"uri, expected_params",
[
("prefix+test://foo/test/123", {"x": "foo", "y": "123"}),
("prefix+test://bar/test/456", {"x": "bar", "y": "456"}),
("test://foo/test/123", None),
("other.prefix+test://foo/test/123", None),
("other+prefix+test://foo/test/123", None),
],
)
def test_match_prefixed_uri_template(
self, uri: str, expected_params: dict[str, str] | None
):
"""Test matching URIs against a template with a prefix."""
uri_template = "prefix+test://{x}/test/{y}"
result = match_uri_template(uri=uri, uri_template=uri_template)
assert result == expected_params