diff --git a/src/mcp/server/fastmcp/resources/base.py b/src/mcp/server/fastmcp/resources/base.py index c733e1a46b..c452836eeb 100644 --- a/src/mcp/server/fastmcp/resources/base.py +++ b/src/mcp/server/fastmcp/resources/base.py @@ -1,7 +1,7 @@ """Base classes and interfaces for FastMCP resources.""" import abc -from typing import Annotated +from typing import Annotated, Any from pydantic import ( AnyUrl, @@ -32,6 +32,7 @@ class Resource(BaseModel, abc.ABC): ) icons: list[Icon] | None = Field(default=None, description="Optional list of icons for this resource") annotations: Annotations | None = Field(default=None, description="Optional annotations for the resource") + meta: dict[str, Any] | None = Field(alias="_meta", default=None, description="Optional metadata for the resource") @field_validator("name", mode="before") @classmethod diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index 865b8e7e72..fcbbac396e 100644 --- a/src/mcp/server/fastmcp/server.py +++ b/src/mcp/server/fastmcp/server.py @@ -373,7 +373,7 @@ async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContent try: content = await resource.read() - return [ReadResourceContents(content=content, mime_type=resource.mime_type)] + return [ReadResourceContents(content=content, mime_type=resource.mime_type, meta=resource.meta)] except Exception as e: # pragma: no cover logger.exception(f"Error reading resource {uri}") raise ResourceError(str(e)) diff --git a/src/mcp/server/lowlevel/helper_types.py b/src/mcp/server/lowlevel/helper_types.py index 3d09b25056..fecc716db6 100644 --- a/src/mcp/server/lowlevel/helper_types.py +++ b/src/mcp/server/lowlevel/helper_types.py @@ -1,4 +1,5 @@ from dataclasses import dataclass +from typing import Any @dataclass @@ -7,3 +8,4 @@ class ReadResourceContents: content: str | bytes mime_type: str | None = None + meta: dict[str, Any] | None = None diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py index 49d289fb75..71eb74f418 100644 --- a/src/mcp/server/lowlevel/server.py +++ b/src/mcp/server/lowlevel/server.py @@ -318,13 +318,14 @@ def decorator( async def handler(req: types.ReadResourceRequest): result = await func(req.params.uri) - def create_content(data: str | bytes, mime_type: str | None): + def create_content(data: str | bytes, mime_type: str | None, meta: dict[str, Any] | None = None): match data: case str() as data: return types.TextResourceContents( uri=req.params.uri, text=data, mimeType=mime_type or "text/plain", + meta=meta, ) case bytes() as data: # pragma: no cover import base64 @@ -333,6 +334,7 @@ def create_content(data: str | bytes, mime_type: str | None): uri=req.params.uri, blob=base64.b64encode(data).decode(), mimeType=mime_type or "application/octet-stream", + meta=meta, ) match result: @@ -346,7 +348,8 @@ def create_content(data: str | bytes, mime_type: str | None): content = create_content(data, None) case Iterable() as contents: contents_list = [ - create_content(content_item.content, content_item.mime_type) for content_item in contents + create_content(content_item.content, content_item.mime_type, content_item.meta) + for content_item in contents ] return types.ServerResult( types.ReadResourceResult( diff --git a/tests/server/fastmcp/resources/test_resource_meta.py b/tests/server/fastmcp/resources/test_resource_meta.py new file mode 100644 index 0000000000..654a817bcf --- /dev/null +++ b/tests/server/fastmcp/resources/test_resource_meta.py @@ -0,0 +1,193 @@ +"""Tests for _meta attribute support in FastMCP resources.""" + +import pytest +from pydantic import AnyUrl + +import mcp.types as types +from mcp.server.fastmcp import FastMCP +from mcp.server.fastmcp.resources import FunctionResource + + +@pytest.mark.anyio +async def test_resource_with_meta_direct_creation(): + """Test resource with _meta attribute via direct creation.""" + mcp = FastMCP() + + def get_data() -> str: + return "data" + + resource = FunctionResource.from_function( + fn=get_data, + uri="resource://test", + **{"_meta": {"widgetDomain": "example.com"}}, + ) + mcp.add_resource(resource) + + # Get the resource + retrieved = await mcp._resource_manager.get_resource("resource://test") + assert retrieved is not None + assert retrieved.meta is not None + assert retrieved.meta["widgetDomain"] == "example.com" + + # Read the resource and verify _meta is passed through + contents = await mcp.read_resource("resource://test") + assert len(contents) == 1 + assert contents[0].meta is not None + assert contents[0].meta["widgetDomain"] == "example.com" + + +@pytest.mark.anyio +async def test_resource_with_meta_from_function(): + """Test creating a resource with _meta using from_function.""" + + def get_data() -> str: + return "data" + + resource = FunctionResource.from_function( + fn=get_data, + uri="resource://test", + **{"_meta": {"custom": "value", "key": 123}}, + ) + + assert resource.meta is not None + assert resource.meta["custom"] == "value" + assert resource.meta["key"] == 123 + + +@pytest.mark.anyio +async def test_resource_without_meta(): + """Test that resources work correctly without _meta (backwards compatibility).""" + mcp = FastMCP() + + @mcp.resource("resource://test") + def get_test() -> str: + """A test resource.""" + return "test data" + + # Get the resource + resource = await mcp._resource_manager.get_resource("resource://test") + assert resource is not None + assert resource.meta is None + + # Read the resource and verify _meta is None + contents = await mcp.read_resource("resource://test") + assert len(contents) == 1 + assert contents[0].meta is None + + +@pytest.mark.anyio +async def test_resource_meta_end_to_end(): + """Test _meta attributes end-to-end with server handler.""" + mcp = FastMCP() + + def get_widget() -> str: + """A widget resource.""" + return "widget content" + + resource = FunctionResource.from_function( + fn=get_widget, + uri="resource://widget", + **{"_meta": {"widgetDomain": "example.com", "version": "1.0"}}, + ) + mcp.add_resource(resource) + + # Simulate the full request/response cycle + # Get the handler + handler = mcp._mcp_server.request_handlers[types.ReadResourceRequest] + + # Create a request + request = types.ReadResourceRequest( + params=types.ReadResourceRequestParams(uri=AnyUrl("resource://widget")), + ) + + # Call the handler + result = await handler(request) + assert isinstance(result.root, types.ReadResourceResult) + assert len(result.root.contents) == 1 + + content = result.root.contents[0] + assert isinstance(content, types.TextResourceContents) + assert content.text == "widget content" + assert content.meta is not None + assert content.meta["widgetDomain"] == "example.com" + assert content.meta["version"] == "1.0" + + +@pytest.mark.anyio +async def test_resource_meta_with_complex_nested_structure(): + """Test _meta with complex nested data structures.""" + mcp = FastMCP() + + complex_meta = { + "widgetDomain": "example.com", + "config": {"nested": {"value": 42}, "list": [1, 2, 3]}, + "tags": ["tag1", "tag2"], + } + + def get_complex() -> str: + """A resource with complex _meta.""" + return "complex data" + + resource = FunctionResource.from_function( + fn=get_complex, + uri="resource://complex", + **{"_meta": complex_meta}, + ) + mcp.add_resource(resource) + + # Read the resource + contents = await mcp.read_resource("resource://complex") + assert len(contents) == 1 + assert contents[0].meta is not None + assert contents[0].meta["widgetDomain"] == "example.com" + assert contents[0].meta["config"]["nested"]["value"] == 42 + assert contents[0].meta["config"]["list"] == [1, 2, 3] + assert contents[0].meta["tags"] == ["tag1", "tag2"] + + +@pytest.mark.anyio +async def test_resource_meta_json_serialization(): + """Test that _meta is correctly serialized as '_meta' in JSON output.""" + mcp = FastMCP() + + def get_widget() -> str: + return "widget content" + + resource = FunctionResource.from_function( + fn=get_widget, + uri="resource://widget", + **{"_meta": {"widgetDomain": "example.com", "version": "1.0"}}, + ) + mcp.add_resource(resource) + + # First check the resource itself serializes correctly + resource_json = resource.model_dump(by_alias=True, mode="json") + assert "_meta" in resource_json, "Expected '_meta' key in resource JSON" + assert resource_json["_meta"]["widgetDomain"] == "example.com" + + # Get the full response through the handler + handler = mcp._mcp_server.request_handlers[types.ReadResourceRequest] + request = types.ReadResourceRequest( + params=types.ReadResourceRequestParams(uri=AnyUrl("resource://widget")), + ) + result = await handler(request) + + # Serialize to JSON with aliases + result_json = result.model_dump(by_alias=True, mode="json") + + # Verify _meta is in the JSON output (not "meta") + content_json = result_json["root"]["contents"][0] + assert "_meta" in content_json, "Expected '_meta' key in content JSON output" + assert "meta" not in content_json or content_json.get("meta") is None, "Should not have 'meta' key in JSON output" + assert content_json["_meta"]["widgetDomain"] == "example.com" + assert content_json["_meta"]["version"] == "1.0" + + # Also verify in the JSON string + result_json_str = result.model_dump_json(by_alias=True) + assert '"_meta"' in result_json_str, "Expected '_meta' string in JSON output" + + # Verify the full structure matches expected MCP format + assert content_json["uri"] == "resource://widget" + assert content_json["text"] == "widget content" + assert content_json["mimeType"] == "text/plain" + assert content_json["_meta"]["widgetDomain"] == "example.com" diff --git a/tests/server/test_resource_meta.py b/tests/server/test_resource_meta.py new file mode 100644 index 0000000000..be58982d74 --- /dev/null +++ b/tests/server/test_resource_meta.py @@ -0,0 +1,223 @@ +"""Tests for _meta attribute support in resources.""" + +from collections.abc import Iterable +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 + + +@pytest.fixture +def temp_file(): + """Create a temporary file for testing.""" + with NamedTemporaryFile(mode="w", delete=False) as f: + f.write("test content") + path = Path(f.name).resolve() + yield path + try: + path.unlink() + except FileNotFoundError: # pragma: no cover + pass + + +@pytest.mark.anyio +async def test_read_resource_text_with_meta(temp_file: Path): + """Test that _meta attributes are passed through for text resources.""" + server = Server("test") + + @server.read_resource() + async def read_resource(uri: AnyUrl) -> Iterable[ReadResourceContents]: + return [ + ReadResourceContents( + content="Hello World", + mime_type="text/plain", + meta={"widgetDomain": "example.com", "custom": "value"}, + ) + ] + + # Get the handler directly from the server + handler = server.request_handlers[types.ReadResourceRequest] + + # Create a request + request = types.ReadResourceRequest( + params=types.ReadResourceRequestParams(uri=FileUrl(temp_file.as_uri())), + ) + + # Call the handler + result = await handler(request) + assert isinstance(result.root, types.ReadResourceResult) + assert len(result.root.contents) == 1 + + content = result.root.contents[0] + assert isinstance(content, types.TextResourceContents) + assert content.text == "Hello World" + assert content.mimeType == "text/plain" + assert content.meta is not None + assert content.meta["widgetDomain"] == "example.com" + assert content.meta["custom"] == "value" + + +@pytest.mark.anyio +async def test_read_resource_binary_with_meta(temp_file: Path): + """Test that _meta attributes are passed through for binary resources.""" + server = Server("test") + + @server.read_resource() + async def read_resource(uri: AnyUrl) -> Iterable[ReadResourceContents]: + return [ + ReadResourceContents( + content=b"Hello World", + mime_type="application/octet-stream", + meta={"encoding": "base64", "size": 11}, + ) + ] + + # Get the handler directly from the server + handler = server.request_handlers[types.ReadResourceRequest] + + # Create a request + request = types.ReadResourceRequest( + params=types.ReadResourceRequestParams(uri=FileUrl(temp_file.as_uri())), + ) + + # Call the handler + result = await handler(request) + assert isinstance(result.root, types.ReadResourceResult) + assert len(result.root.contents) == 1 + + content = result.root.contents[0] + assert isinstance(content, types.BlobResourceContents) + assert content.mimeType == "application/octet-stream" + assert content.meta is not None + assert content.meta["encoding"] == "base64" + assert content.meta["size"] == 11 + + +@pytest.mark.anyio +async def test_read_resource_without_meta(temp_file: Path): + """Test that resources work correctly without _meta (backwards compatibility).""" + server = Server("test") + + @server.read_resource() + async def read_resource(uri: AnyUrl) -> Iterable[ReadResourceContents]: + return [ReadResourceContents(content="Hello World", mime_type="text/plain")] + + # Get the handler directly from the server + handler = server.request_handlers[types.ReadResourceRequest] + + # Create a request + request = types.ReadResourceRequest( + params=types.ReadResourceRequestParams(uri=FileUrl(temp_file.as_uri())), + ) + + # Call the handler + result = await handler(request) + assert isinstance(result.root, types.ReadResourceResult) + assert len(result.root.contents) == 1 + + content = result.root.contents[0] + assert isinstance(content, types.TextResourceContents) + assert content.text == "Hello World" + assert content.mimeType == "text/plain" + assert content.meta is None + + +@pytest.mark.anyio +async def test_read_resource_multiple_contents_with_meta(temp_file: Path): + """Test multiple resource contents with different _meta values.""" + server = Server("test") + + @server.read_resource() + async def read_resource(uri: AnyUrl) -> Iterable[ReadResourceContents]: + return [ + ReadResourceContents( + content="First content", + mime_type="text/plain", + meta={"index": 0, "type": "header"}, + ), + ReadResourceContents( + content="Second content", + mime_type="text/plain", + meta={"index": 1, "type": "body"}, + ), + ] + + # Get the handler directly from the server + handler = server.request_handlers[types.ReadResourceRequest] + + # Create a request + request = types.ReadResourceRequest( + params=types.ReadResourceRequestParams(uri=FileUrl(temp_file.as_uri())), + ) + + # Call the handler + result = await handler(request) + assert isinstance(result.root, types.ReadResourceResult) + assert len(result.root.contents) == 2 + + # Check first content + content0 = result.root.contents[0] + assert isinstance(content0, types.TextResourceContents) + assert content0.text == "First content" + assert content0.meta is not None + assert content0.meta["index"] == 0 + assert content0.meta["type"] == "header" + + # Check second content + content1 = result.root.contents[1] + assert isinstance(content1, types.TextResourceContents) + assert content1.text == "Second content" + assert content1.meta is not None + assert content1.meta["index"] == 1 + assert content1.meta["type"] == "body" + + +@pytest.mark.anyio +async def test_read_resource_meta_json_serialization(temp_file: Path): + """Test that _meta is correctly serialized as '_meta' in JSON output.""" + server = Server("test") + + @server.read_resource() + async def read_resource(uri: AnyUrl) -> Iterable[ReadResourceContents]: + return [ + ReadResourceContents( + content="Test content", + mime_type="text/plain", + meta={"widgetDomain": "example.com", "version": "1.0"}, + ) + ] + + # Get the handler + handler = server.request_handlers[types.ReadResourceRequest] + + # Create a request + request = types.ReadResourceRequest( + params=types.ReadResourceRequestParams(uri=FileUrl(temp_file.as_uri())), + ) + + # Call the handler + result = await handler(request) + + # Serialize to JSON with aliases + result_json = result.model_dump(by_alias=True, mode="json") + + # Verify structure + assert "root" in result_json + assert "contents" in result_json["root"] + assert len(result_json["root"]["contents"]) == 1 + + # Verify _meta is in the JSON output (not "meta") + content_json = result_json["root"]["contents"][0] + assert "_meta" in content_json, "Expected '_meta' key in JSON output" + assert "meta" not in content_json or content_json.get("meta") is None, "Should not have 'meta' key in JSON output" + assert content_json["_meta"]["widgetDomain"] == "example.com" + assert content_json["_meta"]["version"] == "1.0" + + # Also verify in the JSON string + result_json_str = result.model_dump_json(by_alias=True) + assert '"_meta"' in result_json_str, "Expected '_meta' string in JSON output" + assert content_json["_meta"]["widgetDomain"] == "example.com"