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.

Author: claude

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

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))

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

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(

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"]

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"