From 218839fa3c1251f5a9df7411b9b7b190c656c6a8 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Nov 2025 06:32:28 +0000 Subject: [PATCH 1/3] Add support for _meta attributes in resource contents This change enables MCP servers to include metadata in resource content using the _meta field, which is part of the MCP specification. This allows servers to provide additional context about resources, such as domain information or other custom metadata. Changes: - Added meta field to ReadResourceContents helper class - Added _meta field to Resource base class with proper alias - Updated lowlevel server handler to pass through _meta when creating TextResourceContents and BlobResourceContents - Updated FastMCP server to pass resource._meta to ReadResourceContents - Added comprehensive tests for _meta support in both lowlevel and FastMCP resources The implementation maintains backward compatibility - resources without _meta continue to work as before with meta=None. --- src/mcp/server/fastmcp/resources/base.py | 3 +- src/mcp/server/fastmcp/server.py | 2 +- src/mcp/server/lowlevel/helper_types.py | 2 + src/mcp/server/lowlevel/server.py | 7 +- .../fastmcp/resources/test_resource_meta.py | 145 +++++++++++++++ tests/server/test_resource_meta.py | 176 ++++++++++++++++++ 6 files changed, 331 insertions(+), 4 deletions(-) create mode 100644 tests/server/fastmcp/resources/test_resource_meta.py create mode 100644 tests/server/test_resource_meta.py diff --git a/src/mcp/server/fastmcp/resources/base.py b/src/mcp/server/fastmcp/resources/base.py index c733e1a46..c452836ee 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 865b8e7e7..fcbbac396 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 3d09b2505..fecc716db 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 49d289fb7..3490123be 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} if meta is not None else {}, ) 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} if meta is not None else {}, ) 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 000000000..256748348 --- /dev/null +++ b/tests/server/fastmcp/resources/test_resource_meta.py @@ -0,0 +1,145 @@ +"""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"] diff --git a/tests/server/test_resource_meta.py b/tests/server/test_resource_meta.py new file mode 100644 index 000000000..c2b761028 --- /dev/null +++ b/tests/server/test_resource_meta.py @@ -0,0 +1,176 @@ +"""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" From c3117b681389760f10b04fe587e9731213ed42aa Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Nov 2025 07:20:12 +0000 Subject: [PATCH 2/3] Add JSON serialization tests for _meta attributes Adds tests to verify that _meta attributes are correctly serialized with the underscore prefix in JSON output, not as "meta". This ensures the implementation conforms to the MCP specification where metadata fields use the _meta key. Tests verify: - Resource serialization includes "_meta" key in JSON - Response content serialization includes "_meta" key - JSON strings contain the literal "_meta" key - No "meta" key appears in serialized output --- .../fastmcp/resources/test_resource_meta.py | 48 +++++++++++++++++++ tests/server/test_resource_meta.py | 47 ++++++++++++++++++ 2 files changed, 95 insertions(+) diff --git a/tests/server/fastmcp/resources/test_resource_meta.py b/tests/server/fastmcp/resources/test_resource_meta.py index 256748348..654a817bc 100644 --- a/tests/server/fastmcp/resources/test_resource_meta.py +++ b/tests/server/fastmcp/resources/test_resource_meta.py @@ -143,3 +143,51 @@ def get_complex() -> str: 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 index c2b761028..be58982d7 100644 --- a/tests/server/test_resource_meta.py +++ b/tests/server/test_resource_meta.py @@ -174,3 +174,50 @@ async def read_resource(uri: AnyUrl) -> Iterable[ReadResourceContents]: 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" From d74ebee977e40fe8f2c02aa0f323306a4da6915f Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 18 Nov 2025 08:01:49 +0000 Subject: [PATCH 3/3] Fix _meta parameter passing in TextResourceContents and BlobResourceContents Use the field name 'meta' instead of the alias '_meta' when constructing Pydantic model instances. The alias is only used for JSON serialization. This fixes a bug where passing **{'_meta': meta} would cause Pydantic validation errors since '_meta' is an alias, not a field name. --- src/mcp/server/lowlevel/server.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py index 3490123be..71eb74f41 100644 --- a/src/mcp/server/lowlevel/server.py +++ b/src/mcp/server/lowlevel/server.py @@ -325,7 +325,7 @@ def create_content(data: str | bytes, mime_type: str | None, meta: dict[str, Any uri=req.params.uri, text=data, mimeType=mime_type or "text/plain", - **{"_meta": meta} if meta is not None else {}, + meta=meta, ) case bytes() as data: # pragma: no cover import base64 @@ -334,7 +334,7 @@ def create_content(data: str | bytes, mime_type: str | None, meta: dict[str, Any uri=req.params.uri, blob=base64.b64encode(data).decode(), mimeType=mime_type or "application/octet-stream", - **{"_meta": meta} if meta is not None else {}, + meta=meta, ) match result: