60e0ffc959
Upgrade checks / Notify on failure (push) Has been cancelled
Upgrade checks / Close issue on success (push) Has been cancelled
Schema Crash Test / Real-world schema crash test (232K schemas) (push) Has been cancelled
Run static analysis / static_analysis (push) Has been cancelled
Tests / Tests: Python 3.10 on ubuntu-latest (push) Has been cancelled
Tests / Tests: Python 3.13 on ubuntu-latest (push) Has been cancelled
Tests / Tests: Python 3.10 on windows-latest (push) Has been cancelled
Tests / Tests with lowest-direct dependencies (push) Has been cancelled
Tests / MCP conformance tests (push) Has been cancelled
Tests / Integration tests (push) Has been cancelled
Tests / Package install smoke (push) Has been cancelled
Upgrade checks / Static analysis (push) Has been cancelled
Upgrade checks / Tests: Python 3.10 on ubuntu-latest (push) Has been cancelled
Upgrade checks / Tests: Python 3.13 on ubuntu-latest (push) Has been cancelled
Upgrade checks / Tests: Python 3.10 on windows-latest (push) Has been cancelled
Upgrade checks / Integration tests (push) Has been cancelled
Update MCPServerConfig Schema / update-config-schema (push) Has been cancelled
Update SDK Documentation / update-sdk-docs (push) Has been cancelled
1017 lines
35 KiB
Python
1017 lines
35 KiB
Python
import functools
|
|
from urllib.parse import quote
|
|
|
|
import pytest
|
|
from pydantic import BaseModel
|
|
|
|
from fastmcp import Context, FastMCP
|
|
from fastmcp.resources import ResourceTemplate
|
|
from fastmcp.resources.function_resource import FunctionResource
|
|
from fastmcp.resources.template import (
|
|
build_regex,
|
|
expand_uri_template,
|
|
match_uri_template,
|
|
)
|
|
|
|
|
|
class TestResourceTemplate:
|
|
"""Test ResourceTemplate functionality."""
|
|
|
|
def test_template_creation(self):
|
|
"""Test creating a template from a function."""
|
|
|
|
def my_func(key: str, value: int) -> dict:
|
|
return {"key": key, "value": value}
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=my_func,
|
|
uri_template="test://{key}/{value}",
|
|
name="test",
|
|
)
|
|
assert template.uri_template == "test://{key}/{value}"
|
|
assert template.name == "test"
|
|
assert template.mime_type == "text/plain" # default
|
|
|
|
assert template.fn(key="test", value=42) == my_func(key="test", value=42)
|
|
|
|
def test_template_matches(self):
|
|
"""Test matching URIs against a template."""
|
|
|
|
def my_func(key: str, value: int) -> dict:
|
|
return {"key": key, "value": value}
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=my_func,
|
|
uri_template="test://{key}/{value}",
|
|
name="test",
|
|
)
|
|
|
|
# Valid match
|
|
params = template.matches("test://foo/123")
|
|
assert params == {"key": "foo", "value": "123"}
|
|
|
|
# No match
|
|
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."""
|
|
|
|
def my_func() -> dict:
|
|
return {"data": "value"}
|
|
|
|
with pytest.raises(
|
|
ValueError, match="URI template must contain at least one parameter"
|
|
):
|
|
ResourceTemplate.from_function(
|
|
fn=my_func,
|
|
uri_template="test://no-params",
|
|
name="test",
|
|
)
|
|
|
|
def test_template_uri_params_subset_of_function_params(self):
|
|
"""Test validation rule: URI parameters must be a subset of function parameters."""
|
|
|
|
def my_func(key: str, value: int) -> dict:
|
|
return {"key": key, "value": value}
|
|
|
|
# This should work - URI params are a subset of function params
|
|
template = ResourceTemplate.from_function(
|
|
fn=my_func,
|
|
uri_template="test://{key}/{value}",
|
|
name="test",
|
|
)
|
|
assert template.uri_template == "test://{key}/{value}"
|
|
|
|
# This should fail - 'unknown' is not a function parameter
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="Required function arguments .* must be a subset of the URI path parameters",
|
|
):
|
|
ResourceTemplate.from_function(
|
|
fn=my_func,
|
|
uri_template="test://{key}/{unknown}",
|
|
name="test",
|
|
)
|
|
|
|
def test_required_params_subset_of_uri_params(self):
|
|
"""Test validation rule: Required function parameters must be in URI parameters."""
|
|
|
|
# Function with required parameters
|
|
def func_with_required(
|
|
required_param: str, optional_param: str = "default"
|
|
) -> dict:
|
|
return {"required": required_param, "optional": optional_param}
|
|
|
|
# This should work - required param is in URI
|
|
template = ResourceTemplate.from_function(
|
|
fn=func_with_required,
|
|
uri_template="test://{required_param}",
|
|
name="test",
|
|
)
|
|
assert template.uri_template == "test://{required_param}"
|
|
|
|
# This should fail - required param is not in URI
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="Required function arguments .* must be a subset of the URI path parameters",
|
|
):
|
|
ResourceTemplate.from_function(
|
|
fn=func_with_required,
|
|
uri_template="test://{optional_param}",
|
|
name="test",
|
|
)
|
|
|
|
def test_multiple_required_params(self):
|
|
"""Test validation with multiple required parameters."""
|
|
|
|
def multi_required(param1: str, param2: int, optional: str = "default") -> dict:
|
|
return {"p1": param1, "p2": param2, "opt": optional}
|
|
|
|
# This works - all required params in URI
|
|
template = ResourceTemplate.from_function(
|
|
fn=multi_required,
|
|
uri_template="test://{param1}/{param2}",
|
|
name="test",
|
|
)
|
|
assert template.uri_template == "test://{param1}/{param2}"
|
|
|
|
# This fails - missing one required param
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="Required function arguments .* must be a subset of the URI path parameters",
|
|
):
|
|
ResourceTemplate.from_function(
|
|
fn=multi_required,
|
|
uri_template="test://{param1}",
|
|
name="test",
|
|
)
|
|
|
|
async def test_create_resource(self):
|
|
"""Test creating a resource from a template."""
|
|
|
|
def my_func(key: str, value: int) -> str:
|
|
return f"key={key}, value={value}"
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=my_func,
|
|
uri_template="test://{key}/{value}",
|
|
name="test",
|
|
)
|
|
|
|
resource = await template.create_resource(
|
|
"test://foo/123",
|
|
{"key": "foo", "value": 123},
|
|
)
|
|
|
|
assert isinstance(resource, FunctionResource)
|
|
# read() returns raw value from function
|
|
result = await resource.read()
|
|
assert result == "key=foo, value=123"
|
|
|
|
# _read() wraps in ResourceResult
|
|
resource_result = await resource._read()
|
|
assert len(resource_result.contents) == 1
|
|
assert resource_result.contents[0].content == "key=foo, value=123"
|
|
|
|
async def test_async_text_resource(self):
|
|
"""Test creating a text resource from async function."""
|
|
|
|
async def greet(name: str) -> str:
|
|
return f"Hello, {name}!"
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=greet,
|
|
uri_template="greet://{name}",
|
|
name="greeter",
|
|
)
|
|
|
|
resource = await template.create_resource(
|
|
"greet://world",
|
|
{"name": "world"},
|
|
)
|
|
|
|
assert isinstance(resource, FunctionResource)
|
|
# read() returns raw value
|
|
result = await resource.read()
|
|
assert result == "Hello, world!"
|
|
|
|
async def test_async_binary_resource(self):
|
|
"""Test creating a binary resource from async function."""
|
|
|
|
async def get_bytes(value: str) -> bytes:
|
|
return value.encode()
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=get_bytes,
|
|
uri_template="bytes://{value}",
|
|
name="bytes",
|
|
)
|
|
|
|
resource = await template.create_resource(
|
|
"bytes://test",
|
|
{"value": "test"},
|
|
)
|
|
|
|
assert isinstance(resource, FunctionResource)
|
|
# read() returns raw bytes
|
|
result = await resource.read()
|
|
assert result == b"test"
|
|
|
|
async def test_basemodel_conversion(self):
|
|
"""Test handling of BaseModel types."""
|
|
|
|
class MyModel(BaseModel):
|
|
key: str
|
|
value: int
|
|
|
|
def get_data(key: str, value: int) -> MyModel:
|
|
return MyModel(key=key, value=value)
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=get_data,
|
|
uri_template="test://{key}/{value}",
|
|
name="test",
|
|
)
|
|
|
|
resource = await template.create_resource(
|
|
"test://foo/123",
|
|
{"key": "foo", "value": 123},
|
|
)
|
|
|
|
assert isinstance(resource, FunctionResource)
|
|
# read() returns raw BaseModel
|
|
result = await resource.read()
|
|
assert isinstance(result, MyModel)
|
|
data = result.model_dump()
|
|
assert data == {"key": "foo", "value": 123}
|
|
|
|
async def test_custom_type_conversion(self):
|
|
"""Test handling of custom types."""
|
|
|
|
class CustomData:
|
|
def __init__(self, value: str):
|
|
self.value = value
|
|
|
|
def __str__(self) -> str:
|
|
return self.value
|
|
|
|
def get_data(value: str) -> CustomData:
|
|
return CustomData(value)
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=get_data,
|
|
uri_template="test://{value}",
|
|
name="test",
|
|
)
|
|
|
|
resource = await template.create_resource(
|
|
"test://hello",
|
|
{"value": "hello"},
|
|
)
|
|
|
|
assert isinstance(resource, FunctionResource)
|
|
# read() returns raw CustomData object
|
|
result = await resource.read()
|
|
assert isinstance(result, CustomData)
|
|
assert str(result) == "hello"
|
|
|
|
async def test_wildcard_param_can_create_resource(self):
|
|
"""Test that wildcard parameters are valid."""
|
|
|
|
def identity(path: str) -> str:
|
|
return path
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=identity,
|
|
uri_template="test://{path*}.py",
|
|
name="test",
|
|
)
|
|
|
|
assert await template.create_resource(
|
|
"test://path/to/test.py",
|
|
{"path": "path/to/test.py"},
|
|
)
|
|
|
|
async def test_wildcard_param_matches(self):
|
|
def identify(path: str) -> str:
|
|
return path
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=identify,
|
|
uri_template="test://src/{path*}.py",
|
|
name="test",
|
|
)
|
|
# Valid match
|
|
params = template.matches("test://src/path/to/test.py")
|
|
assert params == {"path": "path/to/test"}
|
|
|
|
async def test_multiple_wildcard_params(self):
|
|
"""Test that multiple wildcard parameters are valid."""
|
|
|
|
def identity(path: str, path2: str) -> str:
|
|
return f"{path}/{path2}"
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=identity,
|
|
uri_template="test://{path*}/xyz/{path2*}",
|
|
name="test",
|
|
)
|
|
|
|
params = template.matches("test://path/to/xyz/abc")
|
|
assert params == {"path": "path/to", "path2": "abc"}
|
|
|
|
async def test_wildcard_param_with_regular_param(self):
|
|
"""Test that a wildcard parameter can be used with a regular parameter."""
|
|
|
|
def identity(prefix: str, path: str) -> str:
|
|
return f"{prefix}/{path}"
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=identity,
|
|
uri_template="test://{prefix}/{path*}",
|
|
name="test",
|
|
)
|
|
|
|
params = template.matches("test://src/path/to/test.py")
|
|
assert params == {"prefix": "src", "path": "path/to/test.py"}
|
|
|
|
async def test_function_with_varargs_not_allowed(self):
|
|
def func(x: int, *args: int) -> int:
|
|
return x + sum(args)
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match=r"Functions with \*args are not supported as resource templates",
|
|
):
|
|
ResourceTemplate.from_function(
|
|
fn=func,
|
|
uri_template="test://{x}/{args*}",
|
|
name="test",
|
|
)
|
|
|
|
async def test_function_with_varkwargs_ok(self):
|
|
def func(x: int, **kwargs: int) -> int:
|
|
return x + sum(kwargs.values())
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=func,
|
|
uri_template="test://{x}/{y}/{z}",
|
|
name="test",
|
|
)
|
|
assert template.uri_template == "test://{x}/{y}/{z}"
|
|
|
|
async def test_callable_object_as_template(self):
|
|
"""Test that a callable object can be used as a template."""
|
|
|
|
class MyTemplate:
|
|
"""This is my template"""
|
|
|
|
def __call__(self, x: str) -> str:
|
|
"""ignore this"""
|
|
return f"X was {x}"
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=MyTemplate(),
|
|
uri_template="test://{x}",
|
|
name="test",
|
|
)
|
|
|
|
resource = await template.create_resource(
|
|
"test://foo",
|
|
{"x": "foo"},
|
|
)
|
|
|
|
assert isinstance(resource, FunctionResource)
|
|
# read() returns raw string from __call__
|
|
result = await resource.read()
|
|
assert result == "X was foo"
|
|
|
|
|
|
class TestMatchUriTemplate:
|
|
"""Test match_uri_template function."""
|
|
|
|
@pytest.mark.parametrize(
|
|
"uri, expected_params",
|
|
[
|
|
("test://a/b", None),
|
|
("test://a/b/c", None),
|
|
("test://a/x/b", {"x": "x"}),
|
|
("test://a/x/y/b", None),
|
|
("test://a/1-2/b", {"x": "1-2"}),
|
|
],
|
|
)
|
|
def test_match_uri_template_single_param(
|
|
self, uri: str, expected_params: dict[str, str]
|
|
):
|
|
"""Test that match_uri_template uses the slash delimiter."""
|
|
uri_template = "test://a/{x}/b"
|
|
result = match_uri_template(uri=uri, uri_template=uri_template)
|
|
assert result == expected_params
|
|
|
|
@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"}),
|
|
("test://foo/bar/baz", None),
|
|
("test://foo/email@domain.com", {"x": "foo", "y": "email@domain.com"}),
|
|
("test://two words/foo", {"x": "two words", "y": "foo"}),
|
|
("test://two.words/foo+bar", {"x": "two.words", "y": "foo+bar"}),
|
|
(
|
|
f"test://escaped{quote('/', safe='')}word/bar",
|
|
{"x": "escaped/word", "y": "bar"},
|
|
),
|
|
(
|
|
f"test://escaped{quote('{', safe='')}x{quote('}', safe='')}word/bar",
|
|
{"x": "escaped{x}word", "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_uri_template_with_prefix(
|
|
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
|
|
|
|
def test_match_uri_template_quoted_params(self):
|
|
uri_template = "user://{name}/{email}"
|
|
quoted_name = quote("John Doe", safe="")
|
|
quoted_email = quote("john@example.com", safe="")
|
|
uri = f"user://{quoted_name}/{quoted_email}"
|
|
result = match_uri_template(uri=uri, uri_template=uri_template)
|
|
assert result == {"name": "John Doe", "email": "john@example.com"}
|
|
|
|
@pytest.mark.parametrize(
|
|
"uri, expected_params",
|
|
[
|
|
("test://a/b", None),
|
|
("test://a/b/c", None),
|
|
("test://a/x/b", {"x": "x"}),
|
|
("test://a/x/y/b", {"x": "x/y"}),
|
|
("bad-prefix://a/x/y/b", None),
|
|
("test://a/x/y/z", None),
|
|
],
|
|
)
|
|
def test_match_uri_template_wildcard_param(
|
|
self, uri: str, expected_params: dict[str, str]
|
|
):
|
|
"""Test that match_uri_template uses the slash delimiter."""
|
|
uri_template = "test://a/{x*}/b"
|
|
result = match_uri_template(uri=uri, uri_template=uri_template)
|
|
assert result == expected_params
|
|
|
|
@pytest.mark.parametrize(
|
|
"uri, expected_params",
|
|
[
|
|
("test://a/x/y/b/c/d", {"x": "x/y", "y": "c/d"}),
|
|
("bad-prefix://a/x/y/b/c/d", None),
|
|
("test://a/x/y/c/d", None),
|
|
("test://a/x/b/y", {"x": "x", "y": "y"}),
|
|
],
|
|
)
|
|
def test_match_uri_template_multiple_wildcard_params(
|
|
self, uri: str, expected_params: dict[str, str]
|
|
):
|
|
"""Test that match_uri_template uses the slash delimiter."""
|
|
uri_template = "test://a/{x*}/b/{y*}"
|
|
result = match_uri_template(uri=uri, uri_template=uri_template)
|
|
assert result == expected_params
|
|
|
|
def test_match_uri_template_wildcard_and_literal_param(self):
|
|
"""Test that match_uri_template uses the slash delimiter."""
|
|
uri = "test://a/x/y/b"
|
|
uri_template = "test://a/{x*}/{y}"
|
|
result = match_uri_template(uri=uri, uri_template=uri_template)
|
|
assert result == {"x": "x/y", "y": "b"}
|
|
|
|
def test_match_consecutive_params(self):
|
|
"""Test that consecutive parameters without a / are not matched."""
|
|
uri = "test://a/x/y"
|
|
uri_template = "test://a/{x}{y}"
|
|
result = match_uri_template(uri=uri, uri_template=uri_template)
|
|
assert result is None
|
|
|
|
@pytest.mark.parametrize(
|
|
"uri, expected_params",
|
|
[
|
|
("file://abc/xyz.py", {"path": "xyz"}),
|
|
("file://abc/x/y/z.py", {"path": "x/y/z"}),
|
|
("file://abc/x/y/z/.py", {"path": "x/y/z/"}),
|
|
("file://abc/x/y/z.md", None),
|
|
("file://x/y/z.txt", None),
|
|
],
|
|
)
|
|
def test_match_uri_template_with_non_slash_suffix(
|
|
self, uri: str, expected_params: dict[str, str]
|
|
):
|
|
uri_template = "file://abc/{path*}.py"
|
|
result = match_uri_template(uri=uri, uri_template=uri_template)
|
|
assert result == expected_params
|
|
|
|
@pytest.mark.parametrize(
|
|
"uri, expected_params",
|
|
[
|
|
("resource://test_foo", {"x": "foo"}),
|
|
("resource://test_bar", {"x": "bar"}),
|
|
("resource://test_hello", {"x": "hello"}),
|
|
("resource://test_with_underscores", {"x": "with_underscores"}),
|
|
("resource://test_", None), # Empty parameter not matched
|
|
("resource://test", None), # Missing parameter delimiter
|
|
("resource://other_foo", None), # Wrong prefix
|
|
("other://test_foo", None), # Wrong scheme
|
|
],
|
|
)
|
|
def test_match_uri_template_embedded_param(
|
|
self, uri: str, expected_params: dict[str, str] | None
|
|
):
|
|
"""Test matching URIs where parameter is embedded within a word segment."""
|
|
uri_template = "resource://test_{x}"
|
|
result = match_uri_template(uri=uri, uri_template=uri_template)
|
|
assert result == expected_params
|
|
|
|
@pytest.mark.parametrize(
|
|
"uri, expected_params",
|
|
[
|
|
("resource://prefix_foo_suffix", {"x": "foo"}),
|
|
("resource://prefix_bar_suffix", {"x": "bar"}),
|
|
("resource://prefix_hello_world_suffix", {"x": "hello_world"}),
|
|
("resource://prefix__suffix", None), # Empty parameter not matched
|
|
("resource://prefix_suffix", None), # Missing parameter delimiter
|
|
("resource://other_foo_suffix", None), # Wrong prefix
|
|
("resource://prefix_foo_other", None), # Wrong suffix
|
|
],
|
|
)
|
|
def test_match_uri_template_embedded_param_with_prefix_and_suffix(
|
|
self, uri: str, expected_params: dict[str, str] | None
|
|
):
|
|
"""Test matching URIs where parameter has both prefix and suffix."""
|
|
uri_template = "resource://prefix_{x}_suffix"
|
|
result = match_uri_template(uri=uri, uri_template=uri_template)
|
|
assert result == expected_params
|
|
|
|
|
|
class TestContextHandling:
|
|
"""Test context handling in resource templates."""
|
|
|
|
def test_context_parameter_detection(self):
|
|
"""Test that context parameters are properly detected in
|
|
ResourceTemplate.from_function()."""
|
|
|
|
def template_with_context(x: int, ctx: Context) -> str:
|
|
return str(x)
|
|
|
|
ResourceTemplate.from_function(
|
|
fn=template_with_context,
|
|
uri_template="test://{x}",
|
|
name="test",
|
|
)
|
|
|
|
def template_without_context(x: int) -> str:
|
|
return str(x)
|
|
|
|
ResourceTemplate.from_function(
|
|
fn=template_without_context,
|
|
uri_template="test://{x}",
|
|
name="test",
|
|
)
|
|
|
|
def test_parameterized_context_parameter_detection(self):
|
|
"""Test that parameterized context parameters are properly detected in
|
|
ResourceTemplate.from_function()."""
|
|
|
|
def template_with_context(x: int, ctx: Context) -> str:
|
|
return str(x)
|
|
|
|
ResourceTemplate.from_function(
|
|
fn=template_with_context,
|
|
uri_template="test://{x}",
|
|
name="test",
|
|
)
|
|
|
|
def test_parameterized_union_context_parameter_detection(self):
|
|
"""Test that context parameters in a union are properly detected in
|
|
ResourceTemplate.from_function()."""
|
|
|
|
def template_with_context(x: int, ctx: Context | None) -> str:
|
|
return str(x)
|
|
|
|
ResourceTemplate.from_function(
|
|
fn=template_with_context,
|
|
uri_template="test://{x}",
|
|
name="test",
|
|
)
|
|
|
|
async def test_context_injection(self):
|
|
"""Test that context is properly injected during resource creation."""
|
|
|
|
def resource_with_context(x: int, ctx: Context) -> str:
|
|
assert isinstance(ctx, Context)
|
|
return str(x)
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=resource_with_context,
|
|
uri_template="test://{x}",
|
|
name="test",
|
|
)
|
|
|
|
from fastmcp import FastMCP
|
|
|
|
mcp = FastMCP()
|
|
context = Context(fastmcp=mcp)
|
|
|
|
async with context:
|
|
resource = await template.create_resource(
|
|
"test://42",
|
|
{"x": 42},
|
|
)
|
|
|
|
assert isinstance(resource, FunctionResource)
|
|
# read() returns the raw value
|
|
result = await resource.read()
|
|
assert result == "42"
|
|
|
|
async def test_context_optional(self):
|
|
"""Test that context is optional when creating resources."""
|
|
|
|
def resource_with_context(x: int, ctx: Context | None = None) -> str:
|
|
return str(x)
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=resource_with_context,
|
|
uri_template="test://{x}",
|
|
name="test",
|
|
)
|
|
|
|
# Even for optional context, we need to provide a context
|
|
mcp = FastMCP()
|
|
context = Context(fastmcp=mcp)
|
|
|
|
async with context:
|
|
resource = await template.create_resource(
|
|
"test://42",
|
|
{"x": 42},
|
|
)
|
|
|
|
assert isinstance(resource, FunctionResource)
|
|
# read() returns the raw value
|
|
result = await resource.read()
|
|
assert result == "42"
|
|
|
|
async def test_context_with_functools_wraps_decorator(self):
|
|
"""Regression test for #2524: decorated templates with Context should work."""
|
|
|
|
def custom_decorator(func):
|
|
@functools.wraps(func)
|
|
async def wrapper(*args, **kwargs):
|
|
return await func(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
@custom_decorator
|
|
async def decorated_template(ctx: Context, item_id: int) -> str:
|
|
assert isinstance(ctx, Context)
|
|
return f"item: {item_id}"
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=decorated_template,
|
|
uri_template="test://{item_id}",
|
|
name="test",
|
|
)
|
|
|
|
mcp = FastMCP()
|
|
context = Context(fastmcp=mcp)
|
|
|
|
async with context:
|
|
resource = await template.create_resource("test://42", {"item_id": 42})
|
|
# read() returns the raw value
|
|
result = await resource.read()
|
|
assert result == "item: 42"
|
|
|
|
|
|
class TestMalformedURITemplates:
|
|
"""Test that malformed URI templates from remote servers don't crash."""
|
|
|
|
@pytest.mark.parametrize(
|
|
"template",
|
|
[
|
|
"test://{1leading}/path",
|
|
"test://{123}/path",
|
|
],
|
|
ids=[
|
|
"leading_digit",
|
|
"all_digits",
|
|
],
|
|
)
|
|
def test_build_regex_returns_none_for_invalid_group_names(self, template: str):
|
|
assert build_regex(template) is None
|
|
|
|
def test_build_regex_normalizes_hyphens(self):
|
|
"""Hyphens in param names produce valid regex with underscored groups."""
|
|
regex = build_regex("test://{user-id}/path")
|
|
assert regex is not None
|
|
match = regex.match("test://alice/path")
|
|
assert match is not None
|
|
assert match.group("user_id") == "alice"
|
|
|
|
def test_build_regex_returns_none_for_duplicate_group_names(self):
|
|
assert build_regex("test://{a}/{a}/path") is None
|
|
|
|
@pytest.mark.parametrize(
|
|
"template",
|
|
[
|
|
"test://{a}/{a}/path",
|
|
"test://{1leading}/path",
|
|
],
|
|
ids=[
|
|
"duplicate_groups",
|
|
"leading_digit",
|
|
],
|
|
)
|
|
def test_match_uri_template_returns_none_for_malformed_templates(
|
|
self, template: str
|
|
):
|
|
assert match_uri_template("test://anything/path", template) is None
|
|
|
|
def test_match_uri_template_normalizes_hyphens(self):
|
|
"""Hyphenated params match and return underscored keys."""
|
|
result = match_uri_template("test://alice/path", "test://{user-id}/path")
|
|
assert result == {"user_id": "alice"}
|
|
|
|
def test_resource_template_matches_with_hyphenated_params(self):
|
|
template = ResourceTemplate(
|
|
uri_template="test://{user-id}/path",
|
|
name="test",
|
|
parameters={},
|
|
)
|
|
result = template.matches("test://alice/path")
|
|
assert result == {"user_id": "alice"}
|
|
|
|
async def test_hyphenated_template_end_to_end(self):
|
|
"""Register and read a resource with hyphenated URI param names."""
|
|
from fastmcp import FastMCP
|
|
|
|
mcp = FastMCP("test")
|
|
|
|
@mcp.resource("data://{user-id}/profile")
|
|
def get_profile(user_id: str) -> str:
|
|
return f"profile for {user_id}"
|
|
|
|
templates = await mcp.list_resource_templates()
|
|
assert len(templates) == 1
|
|
assert templates[0].uri_template == "data://{user-id}/profile"
|
|
|
|
result = await mcp.read_resource("data://alice/profile")
|
|
assert "profile for alice" in str(result)
|
|
|
|
def test_build_regex_normalizes_wildcard_hyphens(self):
|
|
"""Wildcard params with hyphens also get underscored groups."""
|
|
regex = build_regex("files://{file-path*}")
|
|
assert regex is not None
|
|
match = regex.match("files://a/b/c")
|
|
assert match is not None
|
|
assert match.group("file_path") == "a/b/c"
|
|
|
|
def test_expand_uri_template_with_hyphens(self):
|
|
"""expand_uri_template maps underscored params to hyphenated placeholders."""
|
|
result = expand_uri_template("data://{user-id}/profile", {"user_id": "alice"})
|
|
assert result == "data://alice/profile"
|
|
|
|
def test_query_param_does_not_clobber_path_param(self):
|
|
"""If path and query have the same normalized name, path wins."""
|
|
result = match_uri_template(
|
|
"test://alice?user-id=bob", "test://{user-id}{?user-id}"
|
|
)
|
|
# path param should be 'alice', not clobbered by query 'bob'
|
|
assert result is not None
|
|
assert result["user_id"] == "alice"
|
|
|
|
def test_query_param_with_blank_value_is_preserved(self):
|
|
"""Regression for #4056: blank query values (e.g. ?format=) should
|
|
not be silently dropped. parse_qs without keep_blank_values=True
|
|
elides empty values, which loses caller intent."""
|
|
result = match_uri_template(
|
|
"resource://data/42?format=", "resource://data/{id}{?format}"
|
|
)
|
|
assert result is not None
|
|
assert result == {"id": "42", "format": ""}
|
|
|
|
def test_query_param_with_blank_and_present_values(self):
|
|
"""Mix of blank and non-blank query values are both surfaced."""
|
|
result = match_uri_template(
|
|
"resource://data/42?format=&verbose=true",
|
|
"resource://data/{id}{?format,verbose}",
|
|
)
|
|
assert result is not None
|
|
assert result == {"id": "42", "format": "", "verbose": "true"}
|
|
|
|
def test_from_function_rejects_hyphen_underscore_collision(self):
|
|
"""Two raw param names that normalize to the same key are rejected."""
|
|
|
|
def handler(user_id: str) -> str:
|
|
return user_id
|
|
|
|
with pytest.raises(ValueError, match="both normalize to 'user_id'"):
|
|
ResourceTemplate.from_function(
|
|
fn=handler,
|
|
uri_template="test://{user-id}/{user_id}",
|
|
name="collision",
|
|
)
|
|
|
|
def test_build_regex_still_works_for_valid_templates(self):
|
|
regex = build_regex("test://{name}/{id}")
|
|
assert regex is not None
|
|
match = regex.match("test://foo/123")
|
|
assert match is not None
|
|
assert match.group("name") == "foo"
|
|
assert match.group("id") == "123"
|
|
|
|
|
|
class TestExpandUriTemplate:
|
|
"""Test expand_uri_template — the inverse of match_uri_template."""
|
|
|
|
@pytest.mark.parametrize(
|
|
"template, params, expected",
|
|
[
|
|
("test://{x}", {"x": "foo"}, "test://foo"),
|
|
("test://{x}/{y}", {"x": "foo", "y": "bar"}, "test://foo/bar"),
|
|
("test://a/{x}/b", {"x": "mid"}, "test://a/mid/b"),
|
|
],
|
|
)
|
|
def test_expand_simple_params(
|
|
self, template: str, params: dict[str, str], expected: str
|
|
):
|
|
assert expand_uri_template(template, params) == expected
|
|
|
|
@pytest.mark.parametrize(
|
|
"template, params, expected",
|
|
[
|
|
("test://{path*}", {"path": "a/b/c"}, "test://a/b/c"),
|
|
("test://{path*}", {"path": "single"}, "test://single"),
|
|
("test://pre/{rest*}", {"rest": "x/y"}, "test://pre/x/y"),
|
|
(
|
|
"test://{a*}/mid/{b*}",
|
|
{"a": "x/y", "b": "p/q"},
|
|
"test://x/y/mid/p/q",
|
|
),
|
|
("test://{x}/{path*}", {"x": "foo", "path": "a/b"}, "test://foo/a/b"),
|
|
],
|
|
)
|
|
def test_expand_wildcard_params(
|
|
self, template: str, params: dict[str, str], expected: str
|
|
):
|
|
assert expand_uri_template(template, params) == expected
|
|
|
|
def test_expand_query_params(self):
|
|
result = expand_uri_template(
|
|
"test://data{?format,verbose}",
|
|
{"format": "json", "verbose": "true"},
|
|
)
|
|
assert result in (
|
|
"test://data?format=json&verbose=true",
|
|
"test://data?verbose=true&format=json",
|
|
)
|
|
|
|
def test_expand_query_params_partial(self):
|
|
result = expand_uri_template(
|
|
"test://data{?format,verbose}",
|
|
{"format": "json"},
|
|
)
|
|
assert result == "test://data?format=json"
|
|
|
|
def test_expand_query_params_none(self):
|
|
result = expand_uri_template("test://data{?format,verbose}", {})
|
|
assert result == "test://data"
|
|
|
|
def test_expand_ignores_extra_params(self):
|
|
result = expand_uri_template("test://{x}", {"x": "foo", "unused": "bar"})
|
|
assert result == "test://foo"
|
|
|
|
@pytest.mark.parametrize(
|
|
"template, params, expected",
|
|
[
|
|
# Simple {var} matches a single segment, so "/" must be encoded.
|
|
(
|
|
"data://{path}/info",
|
|
{"path": "hello/world"},
|
|
"data://hello%2Fworld/info",
|
|
),
|
|
("test://{x}", {"x": "a b"}, "test://a%20b"),
|
|
("test://{x}", {"x": "a?b"}, "test://a%3Fb"),
|
|
("test://{x}", {"x": "a#b"}, "test://a%23b"),
|
|
# Wildcard {var*} may span segments, so "/" is preserved but other
|
|
# reserved characters are still encoded.
|
|
("test://{path*}", {"path": "a/b/c"}, "test://a/b/c"),
|
|
("test://{path*}", {"path": "a b/c"}, "test://a%20b/c"),
|
|
],
|
|
)
|
|
def test_expand_encodes_reserved_characters(
|
|
self, template: str, params: dict[str, str], expected: str
|
|
):
|
|
"""Path values are percent-encoded so the result round-trips through match."""
|
|
assert expand_uri_template(template, params) == expected
|
|
|
|
|
|
class TestMatchExpandRoundTrip:
|
|
"""match_uri_template and expand_uri_template must agree on the template grammar."""
|
|
|
|
@pytest.mark.parametrize(
|
|
"template, uri",
|
|
[
|
|
("test://{x}", "test://foo"),
|
|
("test://{x}/{y}", "test://foo/bar"),
|
|
("test://a/{x}/b", "test://a/mid/b"),
|
|
("test://{path*}", "test://a/b/c"),
|
|
("test://{path*}", "test://single"),
|
|
("test://pre/{rest*}", "test://pre/x/y/z"),
|
|
("test://{x}/{path*}", "test://foo/a/b/c"),
|
|
],
|
|
)
|
|
def test_expand_then_match_is_identity(self, template: str, uri: str):
|
|
"""Extracting params from a URI and expanding them back reproduces the URI."""
|
|
params = match_uri_template(uri, template)
|
|
assert params is not None
|
|
assert expand_uri_template(template, params) == uri
|
|
|
|
@pytest.mark.parametrize(
|
|
"template, params",
|
|
[
|
|
("data://{path}/info", {"path": "hello/world"}),
|
|
("test://{x}", {"x": "a b c"}),
|
|
("test://{x}", {"x": "100%"}),
|
|
("test://{x}/{y}", {"x": "a/b", "y": "c?d"}),
|
|
("test://{path*}", {"path": "a/b/c"}),
|
|
("test://pre/{rest*}", {"rest": "x y/z"}),
|
|
],
|
|
)
|
|
def test_match_then_expand_recovers_params(
|
|
self, template: str, params: dict[str, str]
|
|
):
|
|
"""Expanding params and matching them back reproduces the original values."""
|
|
uri = expand_uri_template(template, params)
|
|
assert match_uri_template(uri, template) == params
|