PrefectHQ · jlowin · Dec 15, 2025 · Dec 13, 2025 · Dec 15, 2025
diff --git a/docs/servers/resources.mdx b/docs/servers/resources.mdx
@@ -138,8 +138,59 @@ FastMCP automatically converts your function's return value into the appropriate
 -   **`str`**: Sent as `TextResourceContents` (with `mime_type="text/plain"` by default).
 -   **`dict`, `list`, `pydantic.BaseModel`**: Automatically serialized to a JSON string and sent as `TextResourceContents` (with `mime_type="application/json"` by default).
 -   **`bytes`**: Base64 encoded and sent as `BlobResourceContents`. You should specify an appropriate `mime_type` (e.g., `"image/png"`, `"application/octet-stream"`).
+-   **`ResourceContent`**: Full control over content, MIME type, and metadata. See [ResourceContent](#resourcecontent) below.
 -   **`None`**: Results in an empty resource content list being returned.
 
+#### ResourceContent
+
+<VersionBadge version="2.14.1" />
+
+For complete control over resource responses, return a `ResourceContent` object. This lets you include metadata alongside your resource content, which is useful for cases like Content Security Policy headers for HTML widgets.
+
+```python
+from fastmcp import FastMCP
+from fastmcp.resources import ResourceContent
+
+mcp = FastMCP(name="WidgetServer")
+
+@mcp.resource("widget://my-widget")
+def get_widget() -> ResourceContent:
+    """Returns an HTML widget with CSP metadata."""
+    return ResourceContent(
+        content="<html><body>My Widget</body></html>",
+        mime_type="text/html",
+        meta={"csp": "script-src 'self'"}
+    )
+```
+
+`ResourceContent` accepts three fields:
+
+**`content`** - The actual resource content. Can be `str` (text content) or `bytes` (binary content). This is the data that will be returned to the client.
+
+**`mime_type`** - Optional MIME type for the content. Defaults to `"text/plain"` for string content and `"application/octet-stream"` for binary content.
+
+**`meta`** - Optional metadata dictionary that will be included in the MCP response's `_meta` field. Use this for runtime metadata like Content Security Policy headers, caching hints, or other client-specific data.
+
+```python
+# Binary content with metadata
+@mcp.resource("images://logo")
+def get_logo() -> ResourceContent:
+    """Returns a logo image with caching metadata."""
+    with open("logo.png", "rb") as f:
+        image_data = f.read()
+    return ResourceContent(
+        content=image_data,
+        mime_type="image/png",
+        meta={"cache-control": "max-age=3600"}
+    )
+```
+
+<Note>
+The `meta` field in `ResourceContent` is for runtime metadata specific to this read response. This is separate from the `meta` parameter in `@mcp.resource(meta={...})`, which provides static metadata about the resource definition itself (returned when listing resources).
+</Note>
+
+You can still return plain `str` or `bytes` from your resource functions—`ResourceContent` is opt-in for when you need to include metadata.
+
 ### Disabling Resources
 
 <VersionBadge version="2.8.0" />

diff --git a/src/fastmcp/resources/__init__.py b/src/fastmcp/resources/__init__.py
@@ -1,4 +1,5 @@
-from .resource import FunctionResource, Resource
+from .resource import FunctionResource, Resource, ResourceContent
+from .resource_manager import ResourceManager
 from .template import ResourceTemplate
 from .types import (
     BinaryResource,
@@ -7,7 +8,6 @@
     HttpResource,
     TextResource,
 )
-from .resource_manager import ResourceManager
 
 __all__ = [
     "BinaryResource",
@@ -16,6 +16,7 @@
     "FunctionResource",
     "HttpResource",
     "Resource",
+    "ResourceContent",
     "ResourceManager",
     "ResourceTemplate",
     "TextResource",

diff --git a/src/fastmcp/resources/resource.py b/src/fastmcp/resources/resource.py
@@ -2,10 +2,14 @@
 
 from __future__ import annotations
 
+import base64
 import inspect
+import warnings
 from collections.abc import Callable
-from typing import TYPE_CHECKING, Annotated, Any
+from typing import Annotated, Any
 
+import mcp.types
+import pydantic
 import pydantic_core
 from mcp.types import Annotations, Icon
 from mcp.types import Resource as SDKResource
@@ -19,15 +23,111 @@
 )
 from typing_extensions import Self
 
+from fastmcp import settings
 from fastmcp.server.dependencies import get_context, without_injected_parameters
 from fastmcp.server.tasks.config import TaskConfig
 from fastmcp.utilities.components import FastMCPComponent
 from fastmcp.utilities.types import (
     get_fn_name,
 )
 
-if TYPE_CHECKING:
-    pass
+
+class ResourceContent(pydantic.BaseModel):
+    """Canonical wrapper for resource content.
+
+    This is the internal representation for all resource reads. Users can
+    return ResourceContent directly for full control, or return simpler types
+    (str, bytes, dict) which will be automatically converted.
+
+    Example:
+        ```python
+        from fastmcp import FastMCP
+        from fastmcp.resources import ResourceContent
+
+        mcp = FastMCP()
+
+        @mcp.resource("widget://my-widget")
+        def my_widget() -> ResourceContent:
+            return ResourceContent(
+                content="<widget html>",
+                meta={"csp": "script-src 'self'"}
+            )
+        ```
+    """
+
+    model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
+
+    content: str | bytes
+    mime_type: str | None = None
+    meta: dict[str, Any] | None = None
+
+    @classmethod
+    def from_value(
+        cls,
+        value: Any,
+        mime_type: str | None = None,
+        meta: dict[str, Any] | None = None,
+    ) -> ResourceContent:
+        """Convert any value to ResourceContent, handling serialization.
+
+        Args:
+            value: The value to convert. Can be:
+                - ResourceContent: returned as-is (meta param ignored)
+                - str: text content
+                - bytes: binary content
+                - other: serialized to JSON string
+
+            mime_type: Optional MIME type override. If not provided:
+                - str → "text/plain"
+                - bytes → "application/octet-stream"
+                - other → "application/json"
+
+            meta: Optional metadata (ignored if value is already ResourceContent)
+
+        Returns:
+            ResourceContent instance
+        """
+        if isinstance(value, ResourceContent):
+            return value
+        if isinstance(value, str):
+            return cls(content=value, mime_type=mime_type or "text/plain", meta=meta)
+        if isinstance(value, bytes):
+            return cls(
+                content=value,
+                mime_type=mime_type or "application/octet-stream",
+                meta=meta,
+            )
+        # dict, list, BaseModel, etc → JSON
+        json_str = pydantic_core.to_json(value, fallback=str).decode()
+        return cls(
+            content=json_str, mime_type=mime_type or "application/json", meta=meta
+        )
+
+    def to_mcp_resource_contents(
+        self, uri: AnyUrl | str
+    ) -> mcp.types.TextResourceContents | mcp.types.BlobResourceContents:
+        """Convert to MCP resource contents type.
+
+        Args:
+            uri: The URI of the resource (required by MCP types)
+
+        Returns:
+            TextResourceContents for str content, BlobResourceContents for bytes
+        """
+        if isinstance(self.content, str):
+            return mcp.types.TextResourceContents(
+                uri=AnyUrl(uri) if isinstance(uri, str) else uri,
+                text=self.content,
+                mimeType=self.mime_type or "text/plain",
+                _meta=self.meta,
+            )
+        else:
+            return mcp.types.BlobResourceContents(
+                uri=AnyUrl(uri) if isinstance(uri, str) else uri,
+                blob=base64.b64encode(self.content).decode(),
+                mimeType=self.mime_type or "application/octet-stream",
+                _meta=self.meta,
+            )
 
 
 class Resource(FastMCPComponent):
@@ -113,14 +213,41 @@ def set_default_name(self) -> Self:
             raise ValueError("Either name or uri must be provided")
         return self
 
-    async def read(self) -> str | bytes:
+    async def read(self) -> str | bytes | ResourceContent:
         """Read the resource content.
 
-        This method is not implemented in the base Resource class and must be
-        implemented by subclasses.
+        This method must be implemented by subclasses. For backwards compatibility,
+        subclasses can return str, bytes, or ResourceContent. However, returning
+        str or bytes is deprecated - new code should return ResourceContent.
+
+        Returns:
+            str | bytes | ResourceContent: The resource content. Returning str
+            or bytes is deprecated; prefer ResourceContent for full control
+            over MIME type and metadata.
         """
         raise NotImplementedError("Subclasses must implement read()")
 
+    async def _read(self) -> ResourceContent:
+        """Internal API that always returns ResourceContent.
+
+        This method calls read() and wraps str/bytes results in ResourceContent.
+        ResourceManager and other internal code should call this method instead
+        of read() directly.
+        """
+        result = await self.read()
+        if isinstance(result, ResourceContent):
+            return result
+        # Deprecated in 2.14.1: returning str/bytes from read()
+        if settings.deprecation_warnings:
+            warnings.warn(
+                f"Resource.read() returning str or bytes is deprecated (since 2.14.1). "
+                f"Return ResourceContent instead. "
+                f"(Resource: {self.__class__.__name__}, URI: {self.uri})",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+        return ResourceContent.from_value(result, mime_type=self.mime_type)
+
     def to_mcp_resource(
         self,
         *,
@@ -224,17 +351,23 @@ def from_function(
             task_config=task_config,
         )
 
-    async def read(self) -> str | bytes:
-        """Read the resource by calling the wrapped function."""
+    async def read(self) -> str | bytes | ResourceContent:
+        """Read the resource by calling the wrapped function.
+
+        Returns:
+            str | bytes | ResourceContent: The resource content. If the user's
+            function returns str, bytes, dict, etc., it will be wrapped
+            in ResourceContent. Nested Resource reads may return raw types.
+        """
         # self.fn is wrapped by without_injected_parameters which handles
         # dependency resolution internally
         result = self.fn()
         if inspect.isawaitable(result):
             result = await result
 
+        # If user returned another Resource, read it recursively
         if isinstance(result, Resource):
             return await result.read()
-        elif isinstance(result, bytes | str):
-            return result
-        else:
-            return pydantic_core.to_json(result, fallback=str).decode()
+
+        # Convert any value to ResourceContent
+        return ResourceContent.from_value(result, mime_type=self.mime_type)
diff --git a/src/fastmcp/resources/resource_manager.py b/src/fastmcp/resources/resource_manager.py
@@ -11,7 +11,7 @@
 
 from fastmcp import settings
 from fastmcp.exceptions import NotFoundError, ResourceError
-from fastmcp.resources.resource import Resource
+from fastmcp.resources.resource import Resource, ResourceContent
 from fastmcp.resources.template import (
     ResourceTemplate,
     match_uri_template,
@@ -286,18 +286,22 @@ async def get_resource(self, uri: AnyUrl | str) -> Resource:
 
         raise NotFoundError(f"Unknown resource: {uri_str}")
 
-    async def read_resource(self, uri: AnyUrl | str) -> str | bytes:
+    async def read_resource(self, uri: AnyUrl | str) -> ResourceContent:
         """
         Internal API for servers: Finds and reads a resource, respecting the
         filtered protocol path.
+
+        Returns:
+            ResourceContent: The canonical content wrapper. All Resource.read()
+            implementations now return ResourceContent.
         """
         uri_str = str(uri)
 
         # 1. Check local resources first. The server will have already applied its filter.
         if uri_str in self._resources:
             resource = await self.get_resource(uri_str)
             try:
-                return await resource.read()
+                return await resource._read()
 
             # raise ResourceErrors as-is
             except ResourceError as e:
@@ -321,7 +325,7 @@ async def read_resource(self, uri: AnyUrl | str) -> str | bytes:
             if (params := match_uri_template(uri_str, key)) is not None:
                 try:
                     resource = await template.create_resource(uri_str, params=params)
-                    return await resource.read()
+                    return await resource._read()
                 except ResourceError as e:
                     logger.exception(
                         f"Error reading resource from template {uri_str!r}"