From eb83a338d41e86339cf34b694cc274a43680acae Mon Sep 17 00:00:00 2001
From: Alexander Reiff <alex.reiff@alltrails.com>
Date: Mon, 13 Oct 2025 14:33:49 -0700
Subject: [PATCH 1/9] feat: Allow ResourceContents objects to be returned
 directly from read_resource handlers
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Updated the low-level server's read_resource decorator to accept TextResourceContents
and BlobResourceContents objects directly, in addition to the existing ReadResourceContents.
This provides more flexibility for resource handlers to construct and return properly
typed ResourceContents objects with full control over all properties.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
---
 src/mcp/server/lowlevel/server.py | 14 ++++++++++++--
 1 file changed, 12 insertions(+), 2 deletions(-)

diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py
index 1a69cbe48..bf4dcc186 100644
--- a/src/mcp/server/lowlevel/server.py
+++ b/src/mcp/server/lowlevel/server.py
@@ -311,7 +311,14 @@ async def handler(_: Any):
 
     def read_resource(self):
         def decorator(
-            func: Callable[[AnyUrl], Awaitable[str | bytes | Iterable[ReadResourceContents]]],
+            func: Callable[
+                [AnyUrl],
+                Awaitable[
+                    str
+                    | bytes
+                    | Iterable[ReadResourceContents | types.TextResourceContents | types.BlobResourceContents]
+                ],
+            ],
         ):
             logger.debug("Registering handler for ReadResourceRequest")
 
@@ -346,7 +353,10 @@ 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
+                            content_item
+                            if isinstance(content_item, types.ResourceContents)
+                            else create_content(content_item.content, content_item.mime_type)
+                            for content_item in contents
                         ]
                         return types.ServerResult(
                             types.ReadResourceResult(

From 94e7668f2bef9f7dcd0a78849fefa26ded14aac7 Mon Sep 17 00:00:00 2001
From: Alexander Reiff <alex.reiff@alltrails.com>
Date: Mon, 13 Oct 2025 14:45:29 -0700
Subject: [PATCH 2/9] feat: Allow FastMCP resources to return ResourceContents
 objects directly
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Updated FastMCP server and resource base classes to support returning
TextResourceContents and BlobResourceContents objects directly from
resource handlers, matching the low-level server functionality.

- Updated Resource.read() abstract method to accept ResourceContents types
- Modified FunctionResource to handle ResourceContents in wrapped functions
- Updated FastMCP server read_resource to pass through ResourceContents
- Updated Context.read_resource to match new return types

This provides FastMCP users with more control over resource properties
and maintains consistency with the low-level server API.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
---
 src/mcp/server/fastmcp/resources/base.py  |  3 ++-
 src/mcp/server/fastmcp/resources/types.py |  5 ++++-
 src/mcp/server/fastmcp/server.py          | 24 +++++++++++++++++++----
 3 files changed, 26 insertions(+), 6 deletions(-)

diff --git a/src/mcp/server/fastmcp/resources/base.py b/src/mcp/server/fastmcp/resources/base.py
index 0bef1a266..141eb8047 100644
--- a/src/mcp/server/fastmcp/resources/base.py
+++ b/src/mcp/server/fastmcp/resources/base.py
@@ -13,6 +13,7 @@
     field_validator,
 )
 
+import mcp.types as types
 from mcp.types import Icon
 
 
@@ -43,6 +44,6 @@ def set_default_name(cls, name: str | None, info: ValidationInfo) -> str:
         raise ValueError("Either name or uri must be provided")
 
     @abc.abstractmethod
-    async def read(self) -> str | bytes:
+    async def read(self) -> str | bytes | types.TextResourceContents | types.BlobResourceContents:
         """Read the resource content."""
         pass
diff --git a/src/mcp/server/fastmcp/resources/types.py b/src/mcp/server/fastmcp/resources/types.py
index c578e23de..14c32ea61 100644
--- a/src/mcp/server/fastmcp/resources/types.py
+++ b/src/mcp/server/fastmcp/resources/types.py
@@ -13,6 +13,7 @@
 import pydantic_core
 from pydantic import AnyUrl, Field, ValidationInfo, validate_call
 
+import mcp.types as types
 from mcp.server.fastmcp.resources.base import Resource
 from mcp.types import Icon
 
@@ -52,7 +53,7 @@ class FunctionResource(Resource):
 
     fn: Callable[[], Any] = Field(exclude=True)
 
-    async def read(self) -> str | bytes:
+    async def read(self) -> str | bytes | types.TextResourceContents | types.BlobResourceContents:
         """Read the resource by calling the wrapped function."""
         try:
             # Call the function first to see if it returns a coroutine
@@ -63,6 +64,8 @@ async def read(self) -> str | bytes:
 
             if isinstance(result, Resource):
                 return await result.read()
+            elif isinstance(result, (types.TextResourceContents, types.BlobResourceContents)):
+                return result
             elif isinstance(result, bytes):
                 return result
             elif isinstance(result, str):
diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py
index 485ef1519..de916d873 100644
--- a/src/mcp/server/fastmcp/server.py
+++ b/src/mcp/server/fastmcp/server.py
@@ -43,7 +43,16 @@
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
 from mcp.shared.context import LifespanContextT, RequestContext, RequestT
-from mcp.types import AnyFunction, ContentBlock, GetPromptResult, Icon, ToolAnnotations
+from mcp.types import (
+    AnyFunction,
+    BlobResourceContents,
+    ContentBlock,
+    GetPromptResult,
+    Icon,
+    ResourceContents,
+    TextResourceContents,
+    ToolAnnotations,
+)
 from mcp.types import Prompt as MCPPrompt
 from mcp.types import PromptArgument as MCPPromptArgument
 from mcp.types import Resource as MCPResource
@@ -340,7 +349,9 @@ async def list_resource_templates(self) -> list[MCPResourceTemplate]:
             for template in templates
         ]
 
-    async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContents]:
+    async def read_resource(
+        self, uri: AnyUrl | str
+    ) -> Iterable[ReadResourceContents | TextResourceContents | BlobResourceContents]:
         """Read a resource by URI."""
 
         context = self.get_context()
@@ -350,7 +361,10 @@ async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContent
 
         try:
             content = await resource.read()
-            return [ReadResourceContents(content=content, mime_type=resource.mime_type)]
+            if isinstance(content, ResourceContents):
+                return [content]
+            else:
+                return [ReadResourceContents(content=content, mime_type=resource.mime_type)]
         except Exception as e:
             logger.exception(f"Error reading resource {uri}")
             raise ResourceError(str(e))
@@ -1124,7 +1138,9 @@ async def report_progress(self, progress: float, total: float | None = None, mes
             message=message,
         )
 
-    async def read_resource(self, uri: str | AnyUrl) -> Iterable[ReadResourceContents]:
+    async def read_resource(
+        self, uri: str | AnyUrl
+    ) -> Iterable[ReadResourceContents | TextResourceContents | BlobResourceContents]:
         """Read a resource by URI.
 
         Args:

From 9fbeaf6dcb7a98848c248ebabfe5ecce427dde79 Mon Sep 17 00:00:00 2001
From: Alexander Reiff <alex.reiff@alltrails.com>
Date: Mon, 13 Oct 2025 15:08:28 -0700
Subject: [PATCH 3/9] test: Add comprehensive tests for ResourceContents direct
 return functionality
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Added test coverage for the new feature allowing resources to return
TextResourceContents and BlobResourceContents objects directly:

Low-level server tests (test_read_resource_direct.py):
- Test direct TextResourceContents return
- Test direct BlobResourceContents return
- Test mixed direct and wrapped content
- Test multiple ResourceContents objects

FastMCP tests (test_resource_contents_direct.py):
- Test custom resources returning ResourceContents
- Test function resources returning ResourceContents
- Test resource templates with ResourceContents
- Test mixed traditional and direct resources

All tests verify proper handling and pass-through of ResourceContents
objects without wrapping them in ReadResourceContents.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
---
 .../test_resource_contents_direct.py          | 190 +++++++++++++++++
 tests/server/test_read_resource_direct.py     | 191 ++++++++++++++++++
 2 files changed, 381 insertions(+)
 create mode 100644 tests/server/fastmcp/resources/test_resource_contents_direct.py
 create mode 100644 tests/server/test_read_resource_direct.py

diff --git a/tests/server/fastmcp/resources/test_resource_contents_direct.py b/tests/server/fastmcp/resources/test_resource_contents_direct.py
new file mode 100644
index 000000000..9f0818d8b
--- /dev/null
+++ b/tests/server/fastmcp/resources/test_resource_contents_direct.py
@@ -0,0 +1,190 @@
+"""Test FastMCP resources returning ResourceContents directly."""
+
+import pytest
+from pydantic import AnyUrl
+
+from mcp.server.fastmcp import FastMCP
+from mcp.server.fastmcp.resources import TextResource
+from mcp.types import BlobResourceContents, TextResourceContents
+
+
+@pytest.mark.anyio
+async def test_resource_returns_text_resource_contents_directly():
+    """Test a custom resource that returns TextResourceContents directly."""
+    app = FastMCP("test")
+
+    class DirectTextResource(TextResource):
+        """A resource that returns TextResourceContents directly."""
+
+        async def read(self):
+            # Return TextResourceContents directly instead of str
+            return TextResourceContents(
+                uri=self.uri,
+                text="Direct TextResourceContents content",
+                mimeType="text/markdown",
+            )
+
+    # Add the resource
+    app.add_resource(
+        DirectTextResource(
+            uri="resource://direct-text",
+            name="direct-text",
+            title="Direct Text Resource",
+            description="Returns TextResourceContents directly",
+            text="This is ignored since we override read()",
+        )
+    )
+
+    # Read the resource
+    contents = await app.read_resource("resource://direct-text")
+    contents_list = list(contents)
+
+    # Verify the result
+    assert len(contents_list) == 1
+    content = contents_list[0]
+    assert isinstance(content, TextResourceContents)
+    assert content.text == "Direct TextResourceContents content"
+    assert content.mimeType == "text/markdown"
+    assert str(content.uri) == "resource://direct-text"
+
+
+@pytest.mark.anyio
+async def test_resource_returns_blob_resource_contents_directly():
+    """Test a custom resource that returns BlobResourceContents directly."""
+    app = FastMCP("test")
+
+    class DirectBlobResource(TextResource):
+        """A resource that returns BlobResourceContents directly."""
+
+        async def read(self):
+            # Return BlobResourceContents directly
+            return BlobResourceContents(
+                uri=self.uri,
+                blob="SGVsbG8gRmFzdE1DUA==",  # "Hello FastMCP" in base64
+                mimeType="application/pdf",
+            )
+
+    # Add the resource
+    app.add_resource(
+        DirectBlobResource(
+            uri="resource://direct-blob",
+            name="direct-blob",
+            title="Direct Blob Resource",
+            description="Returns BlobResourceContents directly",
+            text="This is ignored since we override read()",
+        )
+    )
+
+    # Read the resource
+    contents = await app.read_resource("resource://direct-blob")
+    contents_list = list(contents)
+
+    # Verify the result
+    assert len(contents_list) == 1
+    content = contents_list[0]
+    assert isinstance(content, BlobResourceContents)
+    assert content.blob == "SGVsbG8gRmFzdE1DUA=="
+    assert content.mimeType == "application/pdf"
+    assert str(content.uri) == "resource://direct-blob"
+
+
+@pytest.mark.anyio
+async def test_function_resource_returns_resource_contents():
+    """Test function resource returning ResourceContents directly."""
+    app = FastMCP("test")
+
+    @app.resource("resource://function-text-contents")
+    async def get_text_contents() -> TextResourceContents:
+        """Return TextResourceContents directly from function resource."""
+        return TextResourceContents(
+            uri=AnyUrl("resource://function-text-contents"),
+            text="Function returned TextResourceContents",
+            mimeType="text/x-python",
+        )
+
+    @app.resource("resource://function-blob-contents")
+    def get_blob_contents() -> BlobResourceContents:
+        """Return BlobResourceContents directly from function resource."""
+        return BlobResourceContents(
+            uri=AnyUrl("resource://function-blob-contents"),
+            blob="RnVuY3Rpb24gYmxvYg==",  # "Function blob" in base64
+            mimeType="image/png",
+        )
+
+    # Read text resource
+    text_contents = await app.read_resource("resource://function-text-contents")
+    text_list = list(text_contents)
+    assert len(text_list) == 1
+    text_content = text_list[0]
+    assert isinstance(text_content, TextResourceContents)
+    assert text_content.text == "Function returned TextResourceContents"
+    assert text_content.mimeType == "text/x-python"
+
+    # Read blob resource
+    blob_contents = await app.read_resource("resource://function-blob-contents")
+    blob_list = list(blob_contents)
+    assert len(blob_list) == 1
+    blob_content = blob_list[0]
+    assert isinstance(blob_content, BlobResourceContents)
+    assert blob_content.blob == "RnVuY3Rpb24gYmxvYg=="
+    assert blob_content.mimeType == "image/png"
+
+
+@pytest.mark.anyio
+async def test_mixed_traditional_and_direct_resources():
+    """Test server with both traditional and direct ResourceContents resources."""
+    app = FastMCP("test")
+
+    # Traditional string resource
+    @app.resource("resource://traditional")
+    def traditional_resource() -> str:
+        return "Traditional string content"
+
+    # Direct ResourceContents resource
+    @app.resource("resource://direct")
+    def direct_resource() -> TextResourceContents:
+        return TextResourceContents(
+            uri=AnyUrl("resource://direct"),
+            text="Direct ResourceContents content",
+            mimeType="text/html",
+        )
+
+    # Read traditional resource (will be wrapped)
+    trad_contents = await app.read_resource("resource://traditional")
+    trad_list = list(trad_contents)
+    assert len(trad_list) == 1
+    # The content type might be ReadResourceContents, but we're checking the behavior
+
+    # Read direct ResourceContents
+    direct_contents = await app.read_resource("resource://direct")
+    direct_list = list(direct_contents)
+    assert len(direct_list) == 1
+    direct_content = direct_list[0]
+    assert isinstance(direct_content, TextResourceContents)
+    assert direct_content.text == "Direct ResourceContents content"
+    assert direct_content.mimeType == "text/html"
+
+
+@pytest.mark.anyio
+async def test_resource_template_returns_resource_contents():
+    """Test resource template returning ResourceContents directly."""
+    app = FastMCP("test")
+
+    @app.resource("resource://{category}/{item}")
+    async def get_item_contents(category: str, item: str) -> TextResourceContents:
+        """Return TextResourceContents for template resource."""
+        return TextResourceContents(
+            uri=AnyUrl(f"resource://{category}/{item}"),
+            text=f"Content for {item} in {category}",
+            mimeType="text/plain",
+        )
+
+    # Read templated resource
+    contents = await app.read_resource("resource://books/python")
+    contents_list = list(contents)
+    assert len(contents_list) == 1
+    content = contents_list[0]
+    assert isinstance(content, TextResourceContents)
+    assert content.text == "Content for python in books"
+    assert content.mimeType == "text/plain"
+    assert str(content.uri) == "resource://books/python"
\ No newline at end of file
diff --git a/tests/server/test_read_resource_direct.py b/tests/server/test_read_resource_direct.py
new file mode 100644
index 000000000..b4338f2d1
--- /dev/null
+++ b/tests/server/test_read_resource_direct.py
@@ -0,0 +1,191 @@
+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:
+        pass
+
+
+@pytest.mark.anyio
+async def test_read_resource_direct_text_resource_contents(temp_file: Path):
+    """Test returning TextResourceContents directly from read_resource handler."""
+    server = Server("test")
+
+    @server.read_resource()
+    async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents]:
+        return [
+            types.TextResourceContents(
+                uri=uri,
+                text="Direct text content",
+                mimeType="text/markdown",
+            )
+        ]
+
+    # 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 == "Direct text content"
+    assert content.mimeType == "text/markdown"
+    assert str(content.uri) == temp_file.as_uri()
+
+
+@pytest.mark.anyio
+async def test_read_resource_direct_blob_resource_contents(temp_file: Path):
+    """Test returning BlobResourceContents directly from read_resource handler."""
+    server = Server("test")
+
+    @server.read_resource()
+    async def read_resource(uri: AnyUrl) -> Iterable[types.BlobResourceContents]:
+        return [
+            types.BlobResourceContents(
+                uri=uri,
+                blob="SGVsbG8gV29ybGQ=",  # "Hello World" in base64
+                mimeType="application/pdf",
+            )
+        ]
+
+    # 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.blob == "SGVsbG8gV29ybGQ="
+    assert content.mimeType == "application/pdf"
+    assert str(content.uri) == temp_file.as_uri()
+
+
+@pytest.mark.anyio
+async def test_read_resource_mixed_contents(temp_file: Path):
+    """Test mixing direct ResourceContents with ReadResourceContents."""
+    server = Server("test")
+
+    @server.read_resource()
+    async def read_resource(uri: AnyUrl) -> Iterable[ReadResourceContents | types.TextResourceContents]:
+        return [
+            types.TextResourceContents(
+                uri=uri,
+                text="Direct ResourceContents",
+                mimeType="text/plain",
+            ),
+            ReadResourceContents(content="Wrapped content", mime_type="text/html"),
+        ]
+
+    # 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
+
+    # First content is direct ResourceContents
+    content1 = result.root.contents[0]
+    assert isinstance(content1, types.TextResourceContents)
+    assert content1.text == "Direct ResourceContents"
+    assert content1.mimeType == "text/plain"
+
+    # Second content is wrapped ReadResourceContents
+    content2 = result.root.contents[1]
+    assert isinstance(content2, types.TextResourceContents)
+    assert content2.text == "Wrapped content"
+    assert content2.mimeType == "text/html"
+
+
+@pytest.mark.anyio
+async def test_read_resource_multiple_resource_contents(temp_file: Path):
+    """Test returning multiple ResourceContents objects."""
+    server = Server("test")
+
+    @server.read_resource()
+    async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | types.BlobResourceContents]:
+        return [
+            types.TextResourceContents(
+                uri=uri,
+                text="First text content",
+                mimeType="text/plain",
+            ),
+            types.BlobResourceContents(
+                uri=uri,
+                blob="U2Vjb25kIGNvbnRlbnQ=",  # "Second content" in base64
+                mimeType="application/octet-stream",
+            ),
+            types.TextResourceContents(
+                uri=uri,
+                text="Third text content",
+                mimeType="text/markdown",
+            ),
+        ]
+
+    # 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) == 3
+
+    # Check first content
+    content1 = result.root.contents[0]
+    assert isinstance(content1, types.TextResourceContents)
+    assert content1.text == "First text content"
+    assert content1.mimeType == "text/plain"
+
+    # Check second content
+    content2 = result.root.contents[1]
+    assert isinstance(content2, types.BlobResourceContents)
+    assert content2.blob == "U2Vjb25kIGNvbnRlbnQ="
+    assert content2.mimeType == "application/octet-stream"
+
+    # Check third content
+    content3 = result.root.contents[2]
+    assert isinstance(content3, types.TextResourceContents)
+    assert content3.text == "Third text content"
+    assert content3.mimeType == "text/markdown"
\ No newline at end of file

From 588221a0b2409faa68dfae42150bcb16711674bc Mon Sep 17 00:00:00 2001
From: Alexander Reiff <alex.reiff@alltrails.com>
Date: Mon, 13 Oct 2025 15:16:03 -0700
Subject: [PATCH 4/9] docs: Add examples showing ResourceContents direct return
 with metadata
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Added example snippets demonstrating the main benefit of returning
ResourceContents objects directly - the ability to include metadata
via the _meta field:

- FastMCP example: Shows various metadata use cases including timestamps,
  versions, authorship, image metadata, and query execution details
- Low-level server example: Demonstrates metadata for documents, images,
  multi-part content, and code snippets

The examples emphasize that metadata is the key advantage of this feature,
allowing servers to provide rich contextual information about resources
that helps clients better understand and work with the content.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
---
 .../lowlevel/resource_contents_direct.py      | 267 ++++++++++++++++++
 .../servers/resource_contents_direct.py       | 186 ++++++++++++
 2 files changed, 453 insertions(+)
 create mode 100644 examples/snippets/servers/lowlevel/resource_contents_direct.py
 create mode 100644 examples/snippets/servers/resource_contents_direct.py

diff --git a/examples/snippets/servers/lowlevel/resource_contents_direct.py b/examples/snippets/servers/lowlevel/resource_contents_direct.py
new file mode 100644
index 000000000..0780e93af
--- /dev/null
+++ b/examples/snippets/servers/lowlevel/resource_contents_direct.py
@@ -0,0 +1,267 @@
+"""
+Example showing how to return ResourceContents objects directly from 
+low-level server resources.
+
+The main benefit is the ability to include metadata (_meta field) with
+your resources, providing additional context about the resource content
+such as timestamps, versions, authorship, or any domain-specific metadata.
+"""
+
+import asyncio
+from collections.abc import Iterable
+
+from pydantic import AnyUrl
+
+import mcp.server.stdio as stdio
+import mcp.types as types
+from mcp.server import NotificationOptions, Server
+
+
+# Create a server instance
+server = Server(
+    name="LowLevel ResourceContents Example",
+    version="1.0.0",
+)
+
+
+# Example 1: Return TextResourceContents directly
+@server.read_resource()
+async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | types.BlobResourceContents]:
+    """Handle resource reading with direct ResourceContents return."""
+    uri_str = str(uri)
+    
+    if uri_str == "text://readme":
+        # Return TextResourceContents with document metadata
+        return [
+            types.TextResourceContents(
+                uri=uri,
+                text="# README\n\nThis is a sample readme file.",
+                mimeType="text/markdown",
+                meta={
+                    "title": "Project README",
+                    "author": "Development Team",
+                    "lastModified": "2024-01-15T10:00:00Z",
+                    "version": "2.1.0",
+                    "language": "en",
+                    "license": "MIT",
+                }
+            )
+        ]
+    
+    elif uri_str == "data://config.json":
+        # Return JSON data with schema and validation metadata
+        return [
+            types.TextResourceContents(
+                uri=uri,
+                text='{\n  "version": "1.0.0",\n  "debug": false\n}',
+                mimeType="application/json",
+                meta={
+                    "schema": "https://example.com/schemas/config/v1.0",
+                    "validated": True,
+                    "environment": "production",
+                    "lastValidated": "2024-01-15T14:00:00Z",
+                    "checksum": "sha256:abc123...",
+                }
+            )
+        ]
+    
+    elif uri_str == "image://icon.png":
+        # Return binary data with comprehensive image metadata
+        import base64
+        # This is a 1x1 transparent PNG
+        png_data = base64.b64decode(
+            "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
+        )
+        return [
+            types.BlobResourceContents(
+                uri=uri,
+                blob=base64.b64encode(png_data).decode(),
+                mimeType="image/png",
+                meta={
+                    "width": 1,
+                    "height": 1,
+                    "bitDepth": 8,
+                    "colorType": 6,  # RGBA
+                    "compression": 0,
+                    "filter": 0,
+                    "interlace": 0,
+                    "fileSize": len(png_data),
+                    "hasAlpha": True,
+                    "generated": "2024-01-15T12:00:00Z",
+                    "generator": "Example MCP Server",
+                }
+            )
+        ]
+    
+    elif uri_str == "multi://content":
+        # Return multiple ResourceContents objects with part metadata
+        return [
+            types.TextResourceContents(
+                uri=uri,
+                text="Part 1: Introduction",
+                mimeType="text/plain",
+                meta={
+                    "part": 1,
+                    "title": "Introduction",
+                    "order": 1,
+                    "required": True,
+                }
+            ),
+            types.TextResourceContents(
+                uri=uri,
+                text="## Part 2: Main Content\n\nThis is the main section.",
+                mimeType="text/markdown",
+                meta={
+                    "part": 2,
+                    "title": "Main Content",
+                    "order": 2,
+                    "wordCount": 8,
+                    "headingLevel": 2,
+                }
+            ),
+            types.BlobResourceContents(
+                uri=uri,
+                blob="UGFydCAzOiBCaW5hcnkgRGF0YQ==",  # "Part 3: Binary Data" in base64
+                mimeType="application/octet-stream",
+                meta={
+                    "part": 3,
+                    "title": "Binary Attachment",
+                    "order": 3,
+                    "encoding": "base64",
+                    "originalSize": 19,
+                }
+            ),
+        ]
+    
+    elif uri_str.startswith("code://"):
+        # Extract language from URI for syntax highlighting
+        language = uri_str.split("://")[1].split("/")[0]
+        code_samples = {
+            "python": ('def hello():\n    print("Hello, World!")', "text/x-python"),
+            "javascript": ('console.log("Hello, World!");', "text/javascript"),
+            "html": ('<h1>Hello, World!</h1>', "text/html"),
+        }
+        
+        if language in code_samples:
+            code, mime_type = code_samples[language]
+            return [
+                types.TextResourceContents(
+                    uri=uri,
+                    text=code,
+                    mimeType=mime_type,
+                    meta={
+                        "language": language,
+                        "syntaxHighlighting": True,
+                        "lineNumbers": True,
+                        "executable": language in ["python", "javascript"],
+                        "documentation": f"https://docs.example.com/languages/{language}",
+                    }
+                )
+            ]
+    
+    # Default case - resource not found
+    return [
+        types.TextResourceContents(
+            uri=uri,
+            text=f"Resource not found: {uri}",
+            mimeType="text/plain",
+        )
+    ]
+
+
+# List available resources
+@server.list_resources()
+async def list_resources() -> list[types.Resource]:
+    """List all available resources."""
+    return [
+        types.Resource(
+            uri=AnyUrl("text://readme"),
+            name="README",
+            title="README file",
+            description="A sample readme in markdown format",
+            mimeType="text/markdown",
+        ),
+        types.Resource(
+            uri=AnyUrl("data://config.json"),
+            name="config",
+            title="Configuration",
+            description="Application configuration in JSON format",
+            mimeType="application/json",
+        ),
+        types.Resource(
+            uri=AnyUrl("image://icon.png"),
+            name="icon",
+            title="Application Icon",
+            description="A sample PNG icon",
+            mimeType="image/png",
+        ),
+        types.Resource(
+            uri=AnyUrl("multi://content"),
+            name="multi-part",
+            title="Multi-part Content",
+            description="A resource that returns multiple content items",
+            mimeType="multipart/mixed",
+        ),
+        types.Resource(
+            uri=AnyUrl("code://python/example"),
+            name="python-code",
+            title="Python Code Example",
+            description="Sample Python code with proper MIME type",
+            mimeType="text/x-python",
+        ),
+    ]
+
+
+# Also demonstrate with ReadResourceContents (old style) mixed in
+@server.list_resources()
+async def list_legacy_resources() -> list[types.Resource]:
+    """List resources that use the legacy ReadResourceContents approach."""
+    return [
+        types.Resource(
+            uri=AnyUrl("legacy://text"),
+            name="legacy-text",
+            title="Legacy Text Resource",
+            description="Uses ReadResourceContents wrapper",
+            mimeType="text/plain",
+        ),
+    ]
+
+
+# Mix old and new styles to show compatibility
+from mcp.server.lowlevel.server import ReadResourceContents
+
+
+@server.read_resource()
+async def read_legacy_resource(uri: AnyUrl) -> Iterable[ReadResourceContents | types.TextResourceContents]:
+    """Handle legacy resources alongside new ResourceContents."""
+    uri_str = str(uri)
+    
+    if uri_str == "legacy://text":
+        # Old style - return ReadResourceContents
+        return [
+            ReadResourceContents(
+                content="This uses the legacy ReadResourceContents wrapper",
+                mime_type="text/plain",
+            )
+        ]
+    
+    # Delegate to the new handler for other resources
+    return await read_resource(uri)
+
+
+async def main():
+    """Run the server using stdio transport."""
+    async with stdio.stdio_server() as (read_stream, write_stream):
+        await server.run(
+            read_stream,
+            write_stream,
+            server.create_initialization_options(
+                notification_options=NotificationOptions(),
+                experimental_capabilities={},
+            ),
+        )
+
+
+if __name__ == "__main__":
+    # Run with: python resource_contents_direct.py
+    asyncio.run(main())
\ No newline at end of file
diff --git a/examples/snippets/servers/resource_contents_direct.py b/examples/snippets/servers/resource_contents_direct.py
new file mode 100644
index 000000000..50a5f5da4
--- /dev/null
+++ b/examples/snippets/servers/resource_contents_direct.py
@@ -0,0 +1,186 @@
+"""
+Example showing how to return ResourceContents objects directly from resources.
+
+The main benefit of returning ResourceContents directly is the ability to include
+metadata through the _meta field (exposed as 'meta' in the constructor). This allows
+you to attach additional context to your resources such as:
+
+- Timestamps (created, modified, expires)
+- Version information
+- Author/ownership details
+- File system metadata (permissions, size)
+- Image metadata (dimensions, color space)
+- Document metadata (word count, language)
+- Validation status and schemas
+- Any domain-specific metadata
+
+This metadata helps clients better understand and work with the resource content.
+"""
+
+from mcp.server.fastmcp import FastMCP
+from mcp.types import TextResourceContents, BlobResourceContents
+from pydantic import AnyUrl
+import base64
+
+mcp = FastMCP(name="Direct ResourceContents Example")
+
+
+# Example 1: Return TextResourceContents with metadata
+@mcp.resource("document://report")
+def get_report() -> TextResourceContents:
+    """Return a report with metadata about creation time and author."""
+    return TextResourceContents(
+        uri=AnyUrl("document://report"),
+        text="# Monthly Report\n\nThis is the monthly report content.",
+        mimeType="text/markdown",
+        # The main benefit: adding metadata to the resource
+        meta={
+            "created": "2024-01-15T10:30:00Z",
+            "author": "Analytics Team",
+            "version": "1.2.0",
+            "tags": ["monthly", "finance", "q1-2024"],
+            "confidentiality": "internal",
+        }
+    )
+
+
+# Example 2: Return BlobResourceContents with image metadata
+@mcp.resource("image://logo")
+def get_logo() -> BlobResourceContents:
+    """Return a logo image with metadata about dimensions and format."""
+    # In a real app, you might read this from a file
+    image_bytes = b"\x89PNG\r\n\x1a\n..."  # PNG header
+    
+    return BlobResourceContents(
+        uri=AnyUrl("image://logo"),
+        blob=base64.b64encode(image_bytes).decode(),
+        mimeType="image/png",
+        # Image-specific metadata
+        meta={
+            "width": 512,
+            "height": 512,
+            "format": "PNG",
+            "colorSpace": "sRGB",
+            "hasAlpha": True,
+            "fileSize": 24576,
+            "lastModified": "2024-01-10T08:00:00Z",
+        }
+    )
+
+
+# Example 3: Dynamic resource with real-time metadata
+@mcp.resource("data://metrics/{metric_type}")
+async def get_metrics(metric_type: str) -> TextResourceContents:
+    """Return metrics data with metadata about collection time and source."""
+    import datetime
+    
+    # Simulate collecting metrics
+    metrics = {"cpu": 45.2, "memory": 78.5, "disk": 62.1}
+    timestamp = datetime.datetime.now(datetime.UTC).isoformat()
+    
+    if metric_type == "json":
+        import json
+        return TextResourceContents(
+            uri=AnyUrl(f"data://metrics/{metric_type}"),
+            text=json.dumps(metrics, indent=2),
+            mimeType="application/json",
+            meta={
+                "timestamp": timestamp,
+                "source": "system_monitor",
+                "interval": "5s",
+                "aggregation": "average",
+                "host": "prod-server-01",
+            }
+        )
+    elif metric_type == "csv":
+        csv_text = "metric,value\n" + "\n".join(f"{k},{v}" for k, v in metrics.items())
+        return TextResourceContents(
+            uri=AnyUrl(f"data://metrics/{metric_type}"),
+            text=csv_text,
+            mimeType="text/csv",
+            meta={
+                "timestamp": timestamp,
+                "columns": ["metric", "value"],
+                "row_count": len(metrics),
+            }
+        )
+    else:
+        text = "\n".join(f"{k.upper()}: {v}%" for k, v in metrics.items())
+        return TextResourceContents(
+            uri=AnyUrl(f"data://metrics/{metric_type}"),
+            text=text,
+            mimeType="text/plain",
+            meta={
+                "timestamp": timestamp,
+                "format": "human-readable",
+            }
+        )
+
+
+# Example 4: Configuration resource with version metadata
+@mcp.resource("config://app")
+def get_config() -> TextResourceContents:
+    """Return application config with version and environment metadata."""
+    import json
+    
+    config = {
+        "version": "1.0.0",
+        "features": {
+            "dark_mode": True,
+            "auto_save": False,
+            "language": "en",
+        },
+        "limits": {
+            "max_file_size": 10485760,  # 10MB
+            "max_connections": 100,
+        }
+    }
+    
+    return TextResourceContents(
+        uri=AnyUrl("config://app"),
+        text=json.dumps(config, indent=2),
+        mimeType="application/json",
+        meta={
+            "version": "1.0.0",
+            "lastUpdated": "2024-01-15T14:30:00Z",
+            "environment": "production",
+            "schema": "https://example.com/schemas/config/v1.0",
+            "editable": False,
+        }
+    )
+
+
+# Example 5: Database query result with execution metadata
+@mcp.resource("db://query/users")
+async def get_users() -> TextResourceContents:
+    """Return query results with execution time and row count."""
+    import json
+    import time
+    
+    # Simulate database query
+    start_time = time.time()
+    users = [
+        {"id": 1, "name": "Alice", "role": "admin"},
+        {"id": 2, "name": "Bob", "role": "user"},
+        {"id": 3, "name": "Charlie", "role": "user"},
+    ]
+    execution_time = time.time() - start_time
+    
+    return TextResourceContents(
+        uri=AnyUrl("db://query/users"),
+        text=json.dumps(users, indent=2),
+        mimeType="application/json",
+        meta={
+            "query": "SELECT * FROM users",
+            "executionTime": f"{execution_time:.3f}s",
+            "rowCount": len(users),
+            "database": "main",
+            "cached": False,
+            "timestamp": "2024-01-15T16:00:00Z",
+        }
+    )
+
+
+if __name__ == "__main__":
+    # Run with: python resource_contents_direct.py
+    mcp.run()
\ No newline at end of file

From a6e895076aca0e6f520f6065f93a25b43af9dfe1 Mon Sep 17 00:00:00 2001
From: Alexander Reiff <alex.reiff@alltrails.com>
Date: Mon, 13 Oct 2025 15:22:40 -0700
Subject: [PATCH 5/9] refactor: Clean up imports in FastMCP resource modules
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Updated imports in base.py and types.py to add ResourceContents types
to the existing mcp.types import rather than using a separate import
statement. This follows the existing pattern and keeps imports cleaner.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
---
 src/mcp/server/fastmcp/resources/base.py  | 5 ++---
 src/mcp/server/fastmcp/resources/types.py | 7 +++----
 2 files changed, 5 insertions(+), 7 deletions(-)

diff --git a/src/mcp/server/fastmcp/resources/base.py b/src/mcp/server/fastmcp/resources/base.py
index 141eb8047..349580f9c 100644
--- a/src/mcp/server/fastmcp/resources/base.py
+++ b/src/mcp/server/fastmcp/resources/base.py
@@ -13,8 +13,7 @@
     field_validator,
 )
 
-import mcp.types as types
-from mcp.types import Icon
+from mcp.types import BlobResourceContents, Icon, TextResourceContents
 
 
 class Resource(BaseModel, abc.ABC):
@@ -44,6 +43,6 @@ def set_default_name(cls, name: str | None, info: ValidationInfo) -> str:
         raise ValueError("Either name or uri must be provided")
 
     @abc.abstractmethod
-    async def read(self) -> str | bytes | types.TextResourceContents | types.BlobResourceContents:
+    async def read(self) -> str | bytes | TextResourceContents | BlobResourceContents:
         """Read the resource content."""
         pass
diff --git a/src/mcp/server/fastmcp/resources/types.py b/src/mcp/server/fastmcp/resources/types.py
index 14c32ea61..0f9bd27b6 100644
--- a/src/mcp/server/fastmcp/resources/types.py
+++ b/src/mcp/server/fastmcp/resources/types.py
@@ -13,9 +13,8 @@
 import pydantic_core
 from pydantic import AnyUrl, Field, ValidationInfo, validate_call
 
-import mcp.types as types
 from mcp.server.fastmcp.resources.base import Resource
-from mcp.types import Icon
+from mcp.types import BlobResourceContents, Icon, ResourceContents, TextResourceContents
 
 
 class TextResource(Resource):
@@ -53,7 +52,7 @@ class FunctionResource(Resource):
 
     fn: Callable[[], Any] = Field(exclude=True)
 
-    async def read(self) -> str | bytes | types.TextResourceContents | types.BlobResourceContents:
+    async def read(self) -> str | bytes | TextResourceContents | BlobResourceContents:
         """Read the resource by calling the wrapped function."""
         try:
             # Call the function first to see if it returns a coroutine
@@ -64,7 +63,7 @@ async def read(self) -> str | bytes | types.TextResourceContents | types.BlobRes
 
             if isinstance(result, Resource):
                 return await result.read()
-            elif isinstance(result, (types.TextResourceContents, types.BlobResourceContents)):
+            elif isinstance(result, (TextResourceContents, BlobResourceContents)):
                 return result
             elif isinstance(result, bytes):
                 return result

From 109f881eb674dbdb154d7c757fc3fd0e25f0fed0 Mon Sep 17 00:00:00 2001
From: Alexander Reiff <alex.reiff@alltrails.com>
Date: Mon, 13 Oct 2025 16:25:54 -0700
Subject: [PATCH 6/9] fix: Use _meta field instead of meta in ResourceContents
 examples
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Corrected the examples to use the proper _meta field name as specified
in the MCP protocol, rather than the constructor parameter name 'meta'.
This ensures the metadata appears correctly when viewed in MCP Inspector.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
---
 .../servers/lowlevel/resource_contents_direct.py | 14 +++++++-------
 .../snippets/servers/resource_contents_direct.py | 16 ++++++++--------
 2 files changed, 15 insertions(+), 15 deletions(-)

diff --git a/examples/snippets/servers/lowlevel/resource_contents_direct.py b/examples/snippets/servers/lowlevel/resource_contents_direct.py
index 0780e93af..53a62592a 100644
--- a/examples/snippets/servers/lowlevel/resource_contents_direct.py
+++ b/examples/snippets/servers/lowlevel/resource_contents_direct.py
@@ -37,7 +37,7 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                 uri=uri,
                 text="# README\n\nThis is a sample readme file.",
                 mimeType="text/markdown",
-                meta={
+                _meta={
                     "title": "Project README",
                     "author": "Development Team",
                     "lastModified": "2024-01-15T10:00:00Z",
@@ -55,7 +55,7 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                 uri=uri,
                 text='{\n  "version": "1.0.0",\n  "debug": false\n}',
                 mimeType="application/json",
-                meta={
+                _meta={
                     "schema": "https://example.com/schemas/config/v1.0",
                     "validated": True,
                     "environment": "production",
@@ -77,7 +77,7 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                 uri=uri,
                 blob=base64.b64encode(png_data).decode(),
                 mimeType="image/png",
-                meta={
+                _meta={
                     "width": 1,
                     "height": 1,
                     "bitDepth": 8,
@@ -100,7 +100,7 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                 uri=uri,
                 text="Part 1: Introduction",
                 mimeType="text/plain",
-                meta={
+                _meta={
                     "part": 1,
                     "title": "Introduction",
                     "order": 1,
@@ -111,7 +111,7 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                 uri=uri,
                 text="## Part 2: Main Content\n\nThis is the main section.",
                 mimeType="text/markdown",
-                meta={
+                _meta={
                     "part": 2,
                     "title": "Main Content",
                     "order": 2,
@@ -123,7 +123,7 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                 uri=uri,
                 blob="UGFydCAzOiBCaW5hcnkgRGF0YQ==",  # "Part 3: Binary Data" in base64
                 mimeType="application/octet-stream",
-                meta={
+                _meta={
                     "part": 3,
                     "title": "Binary Attachment",
                     "order": 3,
@@ -149,7 +149,7 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                     uri=uri,
                     text=code,
                     mimeType=mime_type,
-                    meta={
+                    _meta={
                         "language": language,
                         "syntaxHighlighting": True,
                         "lineNumbers": True,
diff --git a/examples/snippets/servers/resource_contents_direct.py b/examples/snippets/servers/resource_contents_direct.py
index 50a5f5da4..ec7813443 100644
--- a/examples/snippets/servers/resource_contents_direct.py
+++ b/examples/snippets/servers/resource_contents_direct.py
@@ -2,7 +2,7 @@
 Example showing how to return ResourceContents objects directly from resources.
 
 The main benefit of returning ResourceContents directly is the ability to include
-metadata through the _meta field (exposed as 'meta' in the constructor). This allows
+metadata through the _meta field. This allows
 you to attach additional context to your resources such as:
 
 - Timestamps (created, modified, expires)
@@ -34,7 +34,7 @@ def get_report() -> TextResourceContents:
         text="# Monthly Report\n\nThis is the monthly report content.",
         mimeType="text/markdown",
         # The main benefit: adding metadata to the resource
-        meta={
+        _meta={
             "created": "2024-01-15T10:30:00Z",
             "author": "Analytics Team",
             "version": "1.2.0",
@@ -56,7 +56,7 @@ def get_logo() -> BlobResourceContents:
         blob=base64.b64encode(image_bytes).decode(),
         mimeType="image/png",
         # Image-specific metadata
-        meta={
+        _meta={
             "width": 512,
             "height": 512,
             "format": "PNG",
@@ -84,7 +84,7 @@ async def get_metrics(metric_type: str) -> TextResourceContents:
             uri=AnyUrl(f"data://metrics/{metric_type}"),
             text=json.dumps(metrics, indent=2),
             mimeType="application/json",
-            meta={
+            _meta={
                 "timestamp": timestamp,
                 "source": "system_monitor",
                 "interval": "5s",
@@ -98,7 +98,7 @@ async def get_metrics(metric_type: str) -> TextResourceContents:
             uri=AnyUrl(f"data://metrics/{metric_type}"),
             text=csv_text,
             mimeType="text/csv",
-            meta={
+            _meta={
                 "timestamp": timestamp,
                 "columns": ["metric", "value"],
                 "row_count": len(metrics),
@@ -110,7 +110,7 @@ async def get_metrics(metric_type: str) -> TextResourceContents:
             uri=AnyUrl(f"data://metrics/{metric_type}"),
             text=text,
             mimeType="text/plain",
-            meta={
+            _meta={
                 "timestamp": timestamp,
                 "format": "human-readable",
             }
@@ -140,7 +140,7 @@ def get_config() -> TextResourceContents:
         uri=AnyUrl("config://app"),
         text=json.dumps(config, indent=2),
         mimeType="application/json",
-        meta={
+        _meta={
             "version": "1.0.0",
             "lastUpdated": "2024-01-15T14:30:00Z",
             "environment": "production",
@@ -170,7 +170,7 @@ async def get_users() -> TextResourceContents:
         uri=AnyUrl("db://query/users"),
         text=json.dumps(users, indent=2),
         mimeType="application/json",
-        meta={
+        _meta={
             "query": "SELECT * FROM users",
             "executionTime": f"{execution_time:.3f}s",
             "rowCount": len(users),

From bd26046e6bf9d87f2f64ccd1b0bdcafb97ae6eb3 Mon Sep 17 00:00:00 2001
From: Alexander Reiff <alex.reiff@alltrails.com>
Date: Mon, 13 Oct 2025 16:33:21 -0700
Subject: [PATCH 7/9] fix: Resolve lint issues and formatting
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Move ReadResourceContents import to top of file in lowlevel example
- Use union syntax (X | Y) instead of tuple in isinstance call
- Fix line length issue by breaking long function signature
- Add proper type annotations for consistency

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
---
 .../lowlevel/resource_contents_direct.py      | 48 +++++++++----------
 .../servers/resource_contents_direct.py       | 41 ++++++++--------
 src/mcp/server/fastmcp/resources/types.py     |  4 +-
 .../test_resource_contents_direct.py          |  2 +-
 tests/server/test_read_resource_direct.py     |  2 +-
 5 files changed, 50 insertions(+), 47 deletions(-)

diff --git a/examples/snippets/servers/lowlevel/resource_contents_direct.py b/examples/snippets/servers/lowlevel/resource_contents_direct.py
index 53a62592a..28298ef0c 100644
--- a/examples/snippets/servers/lowlevel/resource_contents_direct.py
+++ b/examples/snippets/servers/lowlevel/resource_contents_direct.py
@@ -1,5 +1,5 @@
 """
-Example showing how to return ResourceContents objects directly from 
+Example showing how to return ResourceContents objects directly from
 low-level server resources.
 
 The main benefit is the ability to include metadata (_meta field) with
@@ -15,7 +15,7 @@
 import mcp.server.stdio as stdio
 import mcp.types as types
 from mcp.server import NotificationOptions, Server
-
+from mcp.server.lowlevel.server import ReadResourceContents
 
 # Create a server instance
 server = Server(
@@ -29,7 +29,7 @@
 async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | types.BlobResourceContents]:
     """Handle resource reading with direct ResourceContents return."""
     uri_str = str(uri)
-    
+
     if uri_str == "text://readme":
         # Return TextResourceContents with document metadata
         return [
@@ -44,10 +44,10 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                     "version": "2.1.0",
                     "language": "en",
                     "license": "MIT",
-                }
+                },
             )
         ]
-    
+
     elif uri_str == "data://config.json":
         # Return JSON data with schema and validation metadata
         return [
@@ -61,13 +61,14 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                     "environment": "production",
                     "lastValidated": "2024-01-15T14:00:00Z",
                     "checksum": "sha256:abc123...",
-                }
+                },
             )
         ]
-    
+
     elif uri_str == "image://icon.png":
         # Return binary data with comprehensive image metadata
         import base64
+
         # This is a 1x1 transparent PNG
         png_data = base64.b64decode(
             "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
@@ -89,10 +90,10 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                     "hasAlpha": True,
                     "generated": "2024-01-15T12:00:00Z",
                     "generator": "Example MCP Server",
-                }
+                },
             )
         ]
-    
+
     elif uri_str == "multi://content":
         # Return multiple ResourceContents objects with part metadata
         return [
@@ -105,7 +106,7 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                     "title": "Introduction",
                     "order": 1,
                     "required": True,
-                }
+                },
             ),
             types.TextResourceContents(
                 uri=uri,
@@ -117,7 +118,7 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                     "order": 2,
                     "wordCount": 8,
                     "headingLevel": 2,
-                }
+                },
             ),
             types.BlobResourceContents(
                 uri=uri,
@@ -129,19 +130,19 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                     "order": 3,
                     "encoding": "base64",
                     "originalSize": 19,
-                }
+                },
             ),
         ]
-    
+
     elif uri_str.startswith("code://"):
         # Extract language from URI for syntax highlighting
         language = uri_str.split("://")[1].split("/")[0]
         code_samples = {
             "python": ('def hello():\n    print("Hello, World!")', "text/x-python"),
             "javascript": ('console.log("Hello, World!");', "text/javascript"),
-            "html": ('<h1>Hello, World!</h1>', "text/html"),
+            "html": ("<h1>Hello, World!</h1>", "text/html"),
         }
-        
+
         if language in code_samples:
             code, mime_type = code_samples[language]
             return [
@@ -155,10 +156,10 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
                         "lineNumbers": True,
                         "executable": language in ["python", "javascript"],
                         "documentation": f"https://docs.example.com/languages/{language}",
-                    }
+                    },
                 )
             ]
-    
+
     # Default case - resource not found
     return [
         types.TextResourceContents(
@@ -228,14 +229,13 @@ async def list_legacy_resources() -> list[types.Resource]:
 
 
 # Mix old and new styles to show compatibility
-from mcp.server.lowlevel.server import ReadResourceContents
-
-
 @server.read_resource()
-async def read_legacy_resource(uri: AnyUrl) -> Iterable[ReadResourceContents | types.TextResourceContents]:
+async def read_legacy_resource(
+    uri: AnyUrl,
+) -> Iterable[ReadResourceContents | types.TextResourceContents | types.BlobResourceContents]:
     """Handle legacy resources alongside new ResourceContents."""
     uri_str = str(uri)
-    
+
     if uri_str == "legacy://text":
         # Old style - return ReadResourceContents
         return [
@@ -244,7 +244,7 @@ async def read_legacy_resource(uri: AnyUrl) -> Iterable[ReadResourceContents | t
                 mime_type="text/plain",
             )
         ]
-    
+
     # Delegate to the new handler for other resources
     return await read_resource(uri)
 
@@ -264,4 +264,4 @@ async def main():
 
 if __name__ == "__main__":
     # Run with: python resource_contents_direct.py
-    asyncio.run(main())
\ No newline at end of file
+    asyncio.run(main())
diff --git a/examples/snippets/servers/resource_contents_direct.py b/examples/snippets/servers/resource_contents_direct.py
index ec7813443..1f3c10364 100644
--- a/examples/snippets/servers/resource_contents_direct.py
+++ b/examples/snippets/servers/resource_contents_direct.py
@@ -17,11 +17,13 @@
 This metadata helps clients better understand and work with the resource content.
 """
 
-from mcp.server.fastmcp import FastMCP
-from mcp.types import TextResourceContents, BlobResourceContents
-from pydantic import AnyUrl
 import base64
 
+from pydantic import AnyUrl
+
+from mcp.server.fastmcp import FastMCP
+from mcp.types import BlobResourceContents, TextResourceContents
+
 mcp = FastMCP(name="Direct ResourceContents Example")
 
 
@@ -40,7 +42,7 @@ def get_report() -> TextResourceContents:
             "version": "1.2.0",
             "tags": ["monthly", "finance", "q1-2024"],
             "confidentiality": "internal",
-        }
+        },
     )
 
 
@@ -50,7 +52,7 @@ def get_logo() -> BlobResourceContents:
     """Return a logo image with metadata about dimensions and format."""
     # In a real app, you might read this from a file
     image_bytes = b"\x89PNG\r\n\x1a\n..."  # PNG header
-    
+
     return BlobResourceContents(
         uri=AnyUrl("image://logo"),
         blob=base64.b64encode(image_bytes).decode(),
@@ -64,7 +66,7 @@ def get_logo() -> BlobResourceContents:
             "hasAlpha": True,
             "fileSize": 24576,
             "lastModified": "2024-01-10T08:00:00Z",
-        }
+        },
     )
 
 
@@ -73,13 +75,14 @@ def get_logo() -> BlobResourceContents:
 async def get_metrics(metric_type: str) -> TextResourceContents:
     """Return metrics data with metadata about collection time and source."""
     import datetime
-    
+
     # Simulate collecting metrics
     metrics = {"cpu": 45.2, "memory": 78.5, "disk": 62.1}
     timestamp = datetime.datetime.now(datetime.UTC).isoformat()
-    
+
     if metric_type == "json":
         import json
+
         return TextResourceContents(
             uri=AnyUrl(f"data://metrics/{metric_type}"),
             text=json.dumps(metrics, indent=2),
@@ -90,7 +93,7 @@ async def get_metrics(metric_type: str) -> TextResourceContents:
                 "interval": "5s",
                 "aggregation": "average",
                 "host": "prod-server-01",
-            }
+            },
         )
     elif metric_type == "csv":
         csv_text = "metric,value\n" + "\n".join(f"{k},{v}" for k, v in metrics.items())
@@ -102,7 +105,7 @@ async def get_metrics(metric_type: str) -> TextResourceContents:
                 "timestamp": timestamp,
                 "columns": ["metric", "value"],
                 "row_count": len(metrics),
-            }
+            },
         )
     else:
         text = "\n".join(f"{k.upper()}: {v}%" for k, v in metrics.items())
@@ -113,7 +116,7 @@ async def get_metrics(metric_type: str) -> TextResourceContents:
             _meta={
                 "timestamp": timestamp,
                 "format": "human-readable",
-            }
+            },
         )
 
 
@@ -122,7 +125,7 @@ async def get_metrics(metric_type: str) -> TextResourceContents:
 def get_config() -> TextResourceContents:
     """Return application config with version and environment metadata."""
     import json
-    
+
     config = {
         "version": "1.0.0",
         "features": {
@@ -133,9 +136,9 @@ def get_config() -> TextResourceContents:
         "limits": {
             "max_file_size": 10485760,  # 10MB
             "max_connections": 100,
-        }
+        },
     }
-    
+
     return TextResourceContents(
         uri=AnyUrl("config://app"),
         text=json.dumps(config, indent=2),
@@ -146,7 +149,7 @@ def get_config() -> TextResourceContents:
             "environment": "production",
             "schema": "https://example.com/schemas/config/v1.0",
             "editable": False,
-        }
+        },
     )
 
 
@@ -156,7 +159,7 @@ async def get_users() -> TextResourceContents:
     """Return query results with execution time and row count."""
     import json
     import time
-    
+
     # Simulate database query
     start_time = time.time()
     users = [
@@ -165,7 +168,7 @@ async def get_users() -> TextResourceContents:
         {"id": 3, "name": "Charlie", "role": "user"},
     ]
     execution_time = time.time() - start_time
-    
+
     return TextResourceContents(
         uri=AnyUrl("db://query/users"),
         text=json.dumps(users, indent=2),
@@ -177,10 +180,10 @@ async def get_users() -> TextResourceContents:
             "database": "main",
             "cached": False,
             "timestamp": "2024-01-15T16:00:00Z",
-        }
+        },
     )
 
 
 if __name__ == "__main__":
     # Run with: python resource_contents_direct.py
-    mcp.run()
\ No newline at end of file
+    mcp.run()
diff --git a/src/mcp/server/fastmcp/resources/types.py b/src/mcp/server/fastmcp/resources/types.py
index 0f9bd27b6..1d4b1ff53 100644
--- a/src/mcp/server/fastmcp/resources/types.py
+++ b/src/mcp/server/fastmcp/resources/types.py
@@ -14,7 +14,7 @@
 from pydantic import AnyUrl, Field, ValidationInfo, validate_call
 
 from mcp.server.fastmcp.resources.base import Resource
-from mcp.types import BlobResourceContents, Icon, ResourceContents, TextResourceContents
+from mcp.types import BlobResourceContents, Icon, TextResourceContents
 
 
 class TextResource(Resource):
@@ -63,7 +63,7 @@ async def read(self) -> str | bytes | TextResourceContents | BlobResourceContent
 
             if isinstance(result, Resource):
                 return await result.read()
-            elif isinstance(result, (TextResourceContents, BlobResourceContents)):
+            elif isinstance(result, TextResourceContents | BlobResourceContents):
                 return result
             elif isinstance(result, bytes):
                 return result
diff --git a/tests/server/fastmcp/resources/test_resource_contents_direct.py b/tests/server/fastmcp/resources/test_resource_contents_direct.py
index 9f0818d8b..560cd9b51 100644
--- a/tests/server/fastmcp/resources/test_resource_contents_direct.py
+++ b/tests/server/fastmcp/resources/test_resource_contents_direct.py
@@ -187,4 +187,4 @@ async def get_item_contents(category: str, item: str) -> TextResourceContents:
     assert isinstance(content, TextResourceContents)
     assert content.text == "Content for python in books"
     assert content.mimeType == "text/plain"
-    assert str(content.uri) == "resource://books/python"
\ No newline at end of file
+    assert str(content.uri) == "resource://books/python"
diff --git a/tests/server/test_read_resource_direct.py b/tests/server/test_read_resource_direct.py
index b4338f2d1..2a7643b61 100644
--- a/tests/server/test_read_resource_direct.py
+++ b/tests/server/test_read_resource_direct.py
@@ -188,4 +188,4 @@ async def read_resource(uri: AnyUrl) -> Iterable[types.TextResourceContents | ty
     content3 = result.root.contents[2]
     assert isinstance(content3, types.TextResourceContents)
     assert content3.text == "Third text content"
-    assert content3.mimeType == "text/markdown"
\ No newline at end of file
+    assert content3.mimeType == "text/markdown"

From 0ddf851338dd5a6c992637d92ab9d1dd0ff7a8bf Mon Sep 17 00:00:00 2001
From: Alexander Reiff <alex.reiff@alltrails.com>
Date: Mon, 13 Oct 2025 16:49:20 -0700
Subject: [PATCH 8/9] fix: Resolve pyright type errors in ResourceContents
 tests
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Fixed type narrowing issues in test files that were accessing attributes
on union types without proper isinstance checks. The FastMCP read_resource
method returns ReadResourceContents | TextResourceContents | BlobResourceContents,
requiring explicit type checking before accessing type-specific attributes.

Changes:
- Added proper type narrowing with isinstance() checks in all affected tests
- Fixed incorrect Resource base class usage in test files
- Corrected AnyUrl constructor usage in resource creation
- Updated lowlevel example to avoid delegation conflicts

All tests pass and pyright reports 0 errors.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
---
 .../lowlevel/resource_contents_direct.py      | 10 ++++-
 tests/issues/test_141_resource_templates.py   | 17 ++++++++-
 .../test_resource_contents_direct.py          | 12 +++---
 .../fastmcp/servers/test_file_server.py       | 38 +++++++++++++++++--
 tests/server/fastmcp/test_server.py           | 11 +++++-
 5 files changed, 72 insertions(+), 16 deletions(-)

diff --git a/examples/snippets/servers/lowlevel/resource_contents_direct.py b/examples/snippets/servers/lowlevel/resource_contents_direct.py
index 28298ef0c..1d3adcd51 100644
--- a/examples/snippets/servers/lowlevel/resource_contents_direct.py
+++ b/examples/snippets/servers/lowlevel/resource_contents_direct.py
@@ -245,8 +245,14 @@ async def read_legacy_resource(
             )
         ]
 
-    # Delegate to the new handler for other resources
-    return await read_resource(uri)
+    # For other resources, return a simple not found message
+    return [
+        types.TextResourceContents(
+            uri=uri,
+            text=f"Resource not found: {uri}",
+            mimeType="text/plain",
+        )
+    ]
 
 
 async def main():
diff --git a/tests/issues/test_141_resource_templates.py b/tests/issues/test_141_resource_templates.py
index 3145f65e8..23c24b7c3 100644
--- a/tests/issues/test_141_resource_templates.py
+++ b/tests/issues/test_141_resource_templates.py
@@ -53,8 +53,21 @@ def get_user_profile_missing(user_id: str) -> str:
     result = await mcp.read_resource("resource://users/123/posts/456")
     result_list = list(result)
     assert len(result_list) == 1
-    assert result_list[0].content == "Post 456 by user 123"
-    assert result_list[0].mime_type == "text/plain"
+    content = result_list[0]
+    # Since this is a string resource, it should be wrapped as ReadResourceContents
+    from mcp.server.lowlevel.server import ReadResourceContents
+    from mcp.types import TextResourceContents
+
+    if isinstance(content, ReadResourceContents):
+        assert content.content == "Post 456 by user 123"
+        assert content.mime_type == "text/plain"
+    elif isinstance(content, TextResourceContents):
+        # If it's TextResourceContents (direct return)
+        assert content.text == "Post 456 by user 123"
+        assert content.mimeType == "text/plain"
+    else:
+        # Should not happen for string resources
+        raise AssertionError(f"Unexpected content type: {type(content)}")
 
     # Verify invalid parameters raise error
     with pytest.raises(ValueError, match="Unknown resource"):
diff --git a/tests/server/fastmcp/resources/test_resource_contents_direct.py b/tests/server/fastmcp/resources/test_resource_contents_direct.py
index 560cd9b51..5fc4dee2f 100644
--- a/tests/server/fastmcp/resources/test_resource_contents_direct.py
+++ b/tests/server/fastmcp/resources/test_resource_contents_direct.py
@@ -4,7 +4,7 @@
 from pydantic import AnyUrl
 
 from mcp.server.fastmcp import FastMCP
-from mcp.server.fastmcp.resources import TextResource
+from mcp.server.fastmcp.resources import Resource
 from mcp.types import BlobResourceContents, TextResourceContents
 
 
@@ -13,7 +13,7 @@ async def test_resource_returns_text_resource_contents_directly():
     """Test a custom resource that returns TextResourceContents directly."""
     app = FastMCP("test")
 
-    class DirectTextResource(TextResource):
+    class DirectTextResource(Resource):
         """A resource that returns TextResourceContents directly."""
 
         async def read(self):
@@ -27,11 +27,10 @@ async def read(self):
     # Add the resource
     app.add_resource(
         DirectTextResource(
-            uri="resource://direct-text",
+            uri=AnyUrl("resource://direct-text"),
             name="direct-text",
             title="Direct Text Resource",
             description="Returns TextResourceContents directly",
-            text="This is ignored since we override read()",
         )
     )
 
@@ -53,7 +52,7 @@ async def test_resource_returns_blob_resource_contents_directly():
     """Test a custom resource that returns BlobResourceContents directly."""
     app = FastMCP("test")
 
-    class DirectBlobResource(TextResource):
+    class DirectBlobResource(Resource):
         """A resource that returns BlobResourceContents directly."""
 
         async def read(self):
@@ -67,11 +66,10 @@ async def read(self):
     # Add the resource
     app.add_resource(
         DirectBlobResource(
-            uri="resource://direct-blob",
+            uri=AnyUrl("resource://direct-blob"),
             name="direct-blob",
             title="Direct Blob Resource",
             description="Returns BlobResourceContents directly",
-            text="This is ignored since we override read()",
         )
     )
 
diff --git a/tests/server/fastmcp/servers/test_file_server.py b/tests/server/fastmcp/servers/test_file_server.py
index df7024552..ca9653060 100644
--- a/tests/server/fastmcp/servers/test_file_server.py
+++ b/tests/server/fastmcp/servers/test_file_server.py
@@ -92,9 +92,19 @@ async def test_read_resource_dir(mcp: FastMCP):
     res_list = list(res_iter)
     assert len(res_list) == 1
     res = res_list[0]
-    assert res.mime_type == "text/plain"
 
-    files = json.loads(res.content)
+    # Handle union type properly
+    from mcp.server.lowlevel.server import ReadResourceContents
+    from mcp.types import TextResourceContents
+
+    if isinstance(res, ReadResourceContents):
+        assert res.mime_type == "text/plain"
+        files = json.loads(res.content)
+    elif isinstance(res, TextResourceContents):
+        assert res.mimeType == "text/plain"
+        files = json.loads(res.text)
+    else:
+        raise AssertionError(f"Unexpected content type: {type(res)}")
 
     assert sorted([Path(f).name for f in files]) == [
         "config.json",
@@ -109,7 +119,17 @@ async def test_read_resource_file(mcp: FastMCP):
     res_list = list(res_iter)
     assert len(res_list) == 1
     res = res_list[0]
-    assert res.content == "print('hello world')"
+
+    # Handle union type properly
+    from mcp.server.lowlevel.server import ReadResourceContents
+    from mcp.types import TextResourceContents
+
+    if isinstance(res, ReadResourceContents):
+        assert res.content == "print('hello world')"
+    elif isinstance(res, TextResourceContents):
+        assert res.text == "print('hello world')"
+    else:
+        raise AssertionError(f"Unexpected content type: {type(res)}")
 
 
 @pytest.mark.anyio
@@ -125,4 +145,14 @@ async def test_delete_file_and_check_resources(mcp: FastMCP, test_dir: Path):
     res_list = list(res_iter)
     assert len(res_list) == 1
     res = res_list[0]
-    assert res.content == "File not found"
+
+    # Handle union type properly
+    from mcp.server.lowlevel.server import ReadResourceContents
+    from mcp.types import TextResourceContents
+
+    if isinstance(res, ReadResourceContents):
+        assert res.content == "File not found"
+    elif isinstance(res, TextResourceContents):
+        assert res.text == "File not found"
+    else:
+        raise AssertionError(f"Unexpected content type: {type(res)}")
diff --git a/tests/server/fastmcp/test_server.py b/tests/server/fastmcp/test_server.py
index 8caa3b1f6..bd26e16bb 100644
--- a/tests/server/fastmcp/test_server.py
+++ b/tests/server/fastmcp/test_server.py
@@ -1035,7 +1035,16 @@ async def tool_with_resource(ctx: Context[ServerSession, None]) -> str:
             r_list = list(r_iter)
             assert len(r_list) == 1
             r = r_list[0]
-            return f"Read resource: {r.content} with mime type {r.mime_type}"
+            # Handle union type properly
+            from mcp.server.lowlevel.server import ReadResourceContents
+            from mcp.types import TextResourceContents
+
+            if isinstance(r, ReadResourceContents):
+                return f"Read resource: {r.content} with mime type {r.mime_type}"
+            elif isinstance(r, TextResourceContents):
+                return f"Read resource: {r.text} with mime type {r.mimeType}"
+            else:
+                raise AssertionError(f"Unexpected content type: {type(r)}")
 
         async with client_session(mcp._mcp_server) as client:
             result = await client.call_tool("tool_with_resource", {})

From 488b9de7c2dbba805843c854cf541a3b2952a0d8 Mon Sep 17 00:00:00 2001
From: Alexander Reiff <alex.reiff@alltrails.com>
Date: Mon, 13 Oct 2025 16:54:22 -0700
Subject: [PATCH 9/9] fix: Replace datetime.UTC with timezone.utc for Python
 compatibility
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Fixed datetime.UTC usage which is only available in Python 3.11+.
Replaced with datetime.timezone.utc for broader Python version compatibility.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
---
 examples/snippets/servers/resource_contents_direct.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/examples/snippets/servers/resource_contents_direct.py b/examples/snippets/servers/resource_contents_direct.py
index 1f3c10364..ca569aca4 100644
--- a/examples/snippets/servers/resource_contents_direct.py
+++ b/examples/snippets/servers/resource_contents_direct.py
@@ -75,10 +75,11 @@ def get_logo() -> BlobResourceContents:
 async def get_metrics(metric_type: str) -> TextResourceContents:
     """Return metrics data with metadata about collection time and source."""
     import datetime
+    from datetime import timezone
 
     # Simulate collecting metrics
     metrics = {"cpu": 45.2, "memory": 78.5, "disk": 62.1}
-    timestamp = datetime.datetime.now(datetime.UTC).isoformat()
+    timestamp = datetime.datetime.now(timezone.utc).isoformat()
 
     if metric_type == "json":
         import json