Max Isbey
313 lines
9.9 KiB
Python
313 lines
9.9 KiB
Python
import json
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from pydantic import BaseModel
|
|
|
|
from mcp.server.mcpserver import Context, MCPServer
|
|
from mcp.server.mcpserver.resources import FunctionResource, ResourceTemplate
|
|
from mcp.types import Annotations
|
|
|
|
|
|
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[str, Any]:
|
|
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[str, Any]: # pragma: no cover
|
|
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
|
|
|
|
@pytest.mark.anyio
|
|
async def test_create_resource(self):
|
|
"""Test creating a resource from a template."""
|
|
|
|
def my_func(key: str, value: int) -> dict[str, Any]:
|
|
return {"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},
|
|
Context(),
|
|
)
|
|
|
|
assert isinstance(resource, FunctionResource)
|
|
content = await resource.read()
|
|
assert isinstance(content, str)
|
|
data = json.loads(content)
|
|
assert data == {"key": "foo", "value": 123}
|
|
|
|
@pytest.mark.anyio
|
|
async def test_template_error(self):
|
|
"""Test error handling in template resource creation."""
|
|
|
|
def failing_func(x: str) -> str:
|
|
raise ValueError("Test error")
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=failing_func,
|
|
uri_template="fail://{x}",
|
|
name="fail",
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="Error creating resource from template"):
|
|
await template.create_resource("fail://test", {"x": "test"}, Context())
|
|
|
|
@pytest.mark.anyio
|
|
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"},
|
|
Context(),
|
|
)
|
|
|
|
assert isinstance(resource, FunctionResource)
|
|
content = await resource.read()
|
|
assert content == "Hello, world!"
|
|
|
|
@pytest.mark.anyio
|
|
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"},
|
|
Context(),
|
|
)
|
|
|
|
assert isinstance(resource, FunctionResource)
|
|
content = await resource.read()
|
|
assert content == b"test"
|
|
|
|
@pytest.mark.anyio
|
|
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},
|
|
Context(),
|
|
)
|
|
|
|
assert isinstance(resource, FunctionResource)
|
|
content = await resource.read()
|
|
assert isinstance(content, str)
|
|
data = json.loads(content)
|
|
assert data == {"key": "foo", "value": 123}
|
|
|
|
@pytest.mark.anyio
|
|
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"},
|
|
Context(),
|
|
)
|
|
|
|
assert isinstance(resource, FunctionResource)
|
|
content = await resource.read()
|
|
assert content == '"hello"'
|
|
|
|
|
|
class TestResourceTemplateAnnotations:
|
|
"""Test annotations on resource templates."""
|
|
|
|
def test_template_with_annotations(self):
|
|
"""Test creating a template with annotations."""
|
|
|
|
def get_user_data(user_id: str) -> str: # pragma: no cover
|
|
return f"User {user_id}"
|
|
|
|
annotations = Annotations(priority=0.9)
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=get_user_data, uri_template="resource://users/{user_id}", annotations=annotations
|
|
)
|
|
|
|
assert template.annotations is not None
|
|
assert template.annotations.priority == 0.9
|
|
|
|
def test_template_without_annotations(self):
|
|
"""Test that annotations are optional for templates."""
|
|
|
|
def get_user_data(user_id: str) -> str: # pragma: no cover
|
|
return f"User {user_id}"
|
|
|
|
template = ResourceTemplate.from_function(fn=get_user_data, uri_template="resource://users/{user_id}")
|
|
|
|
assert template.annotations is None
|
|
|
|
@pytest.mark.anyio
|
|
async def test_template_annotations_in_mcpserver(self):
|
|
"""Test template annotations via an MCPServer decorator."""
|
|
|
|
mcp = MCPServer()
|
|
|
|
@mcp.resource("resource://dynamic/{id}", annotations=Annotations(audience=["user"], priority=0.7))
|
|
def get_dynamic(id: str) -> str: # pragma: no cover
|
|
"""A dynamic annotated resource."""
|
|
return f"Data for {id}"
|
|
|
|
templates = await mcp.list_resource_templates()
|
|
assert len(templates) == 1
|
|
assert templates[0].annotations is not None
|
|
assert templates[0].annotations.audience == ["user"]
|
|
assert templates[0].annotations.priority == 0.7
|
|
|
|
@pytest.mark.anyio
|
|
async def test_template_created_resources_inherit_annotations(self):
|
|
"""Test that resources created from templates inherit annotations."""
|
|
|
|
def get_item(item_id: str) -> str:
|
|
return f"Item {item_id}"
|
|
|
|
annotations = Annotations(priority=0.6)
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=get_item, uri_template="resource://items/{item_id}", annotations=annotations
|
|
)
|
|
|
|
# Create a resource from the template
|
|
resource = await template.create_resource("resource://items/123", {"item_id": "123"}, Context())
|
|
|
|
# The resource should inherit the template's annotations
|
|
assert resource.annotations is not None
|
|
assert resource.annotations.priority == 0.6
|
|
|
|
# Verify the resource works correctly
|
|
content = await resource.read()
|
|
assert content == "Item 123"
|
|
|
|
|
|
class TestResourceTemplateMetadata:
|
|
"""Test ResourceTemplate meta handling."""
|
|
|
|
def test_template_from_function_with_metadata(self):
|
|
"""Test that ResourceTemplate.from_function() accepts and stores meta parameter."""
|
|
|
|
def get_user(user_id: str) -> str: # pragma: no cover
|
|
return f"User {user_id}"
|
|
|
|
metadata = {"requires_auth": True, "rate_limit": 100}
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=get_user,
|
|
uri_template="resource://users/{user_id}",
|
|
meta=metadata,
|
|
)
|
|
|
|
assert template.meta is not None
|
|
assert template.meta == metadata
|
|
assert template.meta["requires_auth"] is True
|
|
assert template.meta["rate_limit"] == 100
|
|
|
|
@pytest.mark.anyio
|
|
async def test_template_created_resources_inherit_metadata(self):
|
|
"""Test that resources created from templates inherit meta from template."""
|
|
|
|
def get_item(item_id: str) -> str:
|
|
return f"Item {item_id}"
|
|
|
|
metadata = {"category": "inventory", "cacheable": True}
|
|
|
|
template = ResourceTemplate.from_function(
|
|
fn=get_item,
|
|
uri_template="resource://items/{item_id}",
|
|
meta=metadata,
|
|
)
|
|
|
|
# Create a resource from the template
|
|
resource = await template.create_resource("resource://items/123", {"item_id": "123"}, Context())
|
|
|
|
# The resource should inherit the template's metadata
|
|
assert resource.meta is not None
|
|
assert resource.meta == metadata
|
|
assert resource.meta["category"] == "inventory"
|
|
assert resource.meta["cacheable"] is True
|