Merge pull request #843 from gorocode/feature/file-utility-type

Add File utility for wrapping binary data

docs/servers/tools.mdx

@@ -258,6 +258,7 @@ FastMCP automatically converts the value returned by your function into the appr
 - **`bytes`**: Base64 encoded and sent as `BlobResourceContents` (often within an `EmbeddedResource`).
 - **`fastmcp.utilities.types.Image`**: A helper class for easily returning image data. Sent as `ImageContent`.
 - **`fastmcp.utilities.types.Audio`**: A helper class for easily returning audio data. Sent as `AudioContent`.
+- **`fastmcp.utilities.types.File`**: A helper class for easily returning binary data as base64-encoded content. Sent as `EmbeddedResource`.
 - **A list of any of the above**: Automatically converts each item appropriately.
 - **`None`**: Results in an empty response (no content is sent back to the client).
 

examples/get_file.py

@@ -0,0 +1,33 @@
+import aiohttp
+
+from fastmcp.server import FastMCP
+from fastmcp.utilities.types import File
+
+
+def create_server():
+    mcp = FastMCP(name="File Demo", instructions="Get files from the server or URL.")
+
+    @mcp.tool()
+    async def get_test_file_from_server(path: str = "requirements.txt") -> File:
+        """
+        Get a test file from the server. If the path is not provided, it defaults to 'requirements.txt'.
+        """
+        return File(path=path)
+
+    @mcp.tool()
+    async def get_test_pdf_from_url(
+        url: str = "https://mozilla.github.io/pdf.js/web/compressed.tracemonkey-pldi-09.pdf",
+    ) -> File:
+        """
+        Get a test PDF file from a URL. If the URL is not provided, it defaults to a sample PDF.
+        """
+        async with aiohttp.ClientSession() as session:
+            async with session.get(url) as response:
+                pdf_data = await response.read()
+                return File(data=pdf_data, format="pdf")
+
+    return mcp
+
+
+if __name__ == "__main__":
+    create_server().run(transport="sse", host="0.0.0.0", port=8001, path="/sse")

src/fastmcp/tools/tool.py

@@ -18,6 +18,7 @@ from fastmcp.utilities.json_schema import compress_schema
 from fastmcp.utilities.logging import get_logger
 from fastmcp.utilities.types import (
     Audio,
+    File,
     Image,
     MCPContent,
     find_kwarg_by_type,
@@ -277,6 +278,9 @@ def _convert_to_content(
     elif isinstance(result, Audio):
         return [result.to_audio_content()]
 
+    elif isinstance(result, File):
+        return [result.to_resource_content()]
+
     if isinstance(result, list | tuple) and not _process_as_single_item:
         # if the result is a list, then it could either be a list of MCP types,
         # or a "regular" list that the tool is returning, or a mix of both.
@@ -288,7 +292,7 @@ def _convert_to_content(
         other_content = []
 
         for item in result:
-            if isinstance(item, MCPContent | Image | Audio):
+            if isinstance(item, MCPContent | Image | Audio | File):
                 mcp_types.append(_convert_to_content(item)[0])
             else:
                 other_content.append(item)

src/fastmcp/utilities/types.py

@@ -2,6 +2,7 @@
 
 import base64
 import inspect
+import mimetypes
 from collections.abc import Callable
 from functools import lru_cache
 from pathlib import Path
@@ -11,11 +12,13 @@ from typing import Annotated, TypeAlias, TypeVar, Union, get_args, get_origin
 from mcp.types import (
     Annotations,
     AudioContent,
+    BlobResourceContents,
     EmbeddedResource,
     ImageContent,
     TextContent,
+    TextResourceContents,  # Added import
 )
-from pydantic import BaseModel, ConfigDict, TypeAdapter
+from pydantic import AnyUrl, BaseModel, ConfigDict, TypeAdapter, UrlConstraints
 
 T = TypeVar("T")
 
@@ -203,3 +206,89 @@ class Audio:
             mimeType=mime_type or self._mime_type,
             annotations=annotations or self.annotations,
         )
+
+
+class File:
+    """Helper class for returning audio from tools."""
+
+    def __init__(
+        self,
+        path: str | Path | None = None,
+        data: bytes | None = None,
+        format: str | None = None,
+        name: str | None = None,
+        annotations: Annotations | None = None,
+    ):
+        if path is None and data is None:
+            raise ValueError("Either path or data must be provided")
+        if path is not None and data is not None:
+            raise ValueError("Only one of path or data can be provided")
+
+        self.path = Path(path) if path else None
+        self.data = data
+        self._format = format
+        self._mime_type = self._get_mime_type()
+        self._name = name
+        self.annotations = annotations
+
+    def _get_mime_type(self) -> str:
+        """Get MIME type from format or guess from file extension."""
+        if self._format:
+            fmt = self._format.lower()
+            # Map common text formats to text/plain
+            if fmt in {"plain", "txt", "text"}:
+                return "text/plain"
+            return f"application/{fmt}"
+
+        if self.path:
+            mime_type, _ = mimetypes.guess_type(self.path)
+            if mime_type:
+                return mime_type
+
+        return "application/octet-stream"
+
+    def to_resource_content(
+        self,
+        mime_type: str | None = None,
+        annotations: Annotations | None = None,
+    ) -> EmbeddedResource:
+        if self.path:
+            with open(self.path, "rb") as f:
+                raw_data = f.read()
+                uri_str = self.path.resolve().as_uri()
+        elif self.data is not None:
+            raw_data = self.data
+            if self._name:
+                uri_str = f"file:///{self._name}.{self._mime_type.split('/')[1]}"
+            else:
+                uri_str = f"file:///resource.{self._mime_type.split('/')[1]}"
+        else:
+            raise ValueError("No resource data available")
+
+        mime = mime_type or self._mime_type
+        UriType = Annotated[AnyUrl, UrlConstraints(host_required=False)]
+        uri = TypeAdapter(UriType).validate_python(uri_str)
+
+        if mime.startswith("text/"):
+            try:
+                text = raw_data.decode("utf-8")
+            except UnicodeDecodeError:
+                text = raw_data.decode("latin-1")
+            resource = TextResourceContents(
+                text=text,
+                mimeType=mime,
+                uri=uri,
+            )
+        else:
+            data = base64.b64encode(raw_data).decode()
+            resource = BlobResourceContents(
+                blob=data,
+                mimeType=mime,
+                uri=uri,
+            )
+
+        return EmbeddedResource(
+            type="resource",
+            resource=resource,
+            annotations=annotations or self.annotations,
+        )

tests/server/test_server_interactions.py

@@ -11,6 +11,7 @@ import pytest
 from mcp import McpError
 from mcp.types import (
     AudioContent,
+    BlobResourceContents,
     EmbeddedResource,
     ImageContent,
     TextContent,
@@ -25,7 +26,7 @@ from fastmcp.prompts.prompt import Prompt, PromptMessage
 from fastmcp.resources import FileResource, ResourceTemplate
 from fastmcp.resources.resource import FunctionResource
 from fastmcp.tools.tool import Tool
-from fastmcp.utilities.types import Audio, Image
+from fastmcp.utilities.types import Audio, File, Image
 
 
 @pytest.fixture
@@ -53,10 +54,22 @@ def tool_server():
         return Audio(path)
 
     @mcp.tool
-    def mixed_content_tool() -> list[TextContent | ImageContent]:
+    def file_tool(path: str) -> File:
+        return File(path)
+
+    @mcp.tool
+    def mixed_content_tool() -> list[TextContent | ImageContent | EmbeddedResource]:
         return [
             TextContent(type="text", text="Hello"),
-            ImageContent(type="image", data="abc", mimeType="image/png"),
+            ImageContent(type="image", data="abc", mimeType="application/octet-stream"),
+            EmbeddedResource(
+                type="resource",
+                resource=BlobResourceContents(
+                    blob=base64.b64encode(b"abc").decode(),
+                    mimeType="application/octet-stream",
+                    uri=AnyUrl("file:///test.bin"),
+                ),
+            ),
         ]
 
     @mcp.tool
@@ -77,6 +90,20 @@ def tool_server():
             TextContent(type="text", text="direct content"),
         ]
 
+    @mcp.tool
+    def mixed_file_list_fn(file_path: str) -> list:
+        return [
+            "text message",
+            File(file_path),
+            {"key": "value"},
+            TextContent(type="text", text="direct content"),
+        ]
+
+    @mcp.tool
+    def file_text_tool() -> File:
+        # Return a File with text data and text/plain format
+        return File(data=b"hello world", format="plain")
+
     return mcp
 
 
@@ -88,7 +115,7 @@ class TestTools:
 
     async def test_list_tools(self, tool_server: FastMCP):
         async with Client(tool_server) as client:
-            assert len(await client.list_tools()) == 8
+            assert len(await client.list_tools()) == 11
 
     async def test_call_tool(self, tool_server: FastMCP):
         async with Client(tool_server) as client:
@@ -129,6 +156,17 @@ class TestTools:
             result = await client.call_tool("list_tool", {})
             assert result[0].text == '[\n  "x",\n  2\n]'  # type: ignore[attr-defined]
 
+    async def test_file_text_tool(self, tool_server: FastMCP):
+        async with Client(tool_server) as client:
+            result = await client.call_tool("file_text_tool", {})
+            assert len(result) == 1
+            embedded = result[0]
+            assert isinstance(embedded, EmbeddedResource)
+            resource = embedded.resource
+            assert isinstance(resource, TextResourceContents)
+            assert resource.mimeType == "text/plain"
+            assert resource.text == "hello world"
+
 
 class TestToolTags:
     def create_server(self, include_tags=None, exclude_tags=None):
@@ -304,17 +342,52 @@ class TestToolReturnTypes:
             decoded = base64.b64decode(content.data)
             assert decoded == b"fake wav data"
 
+    async def test_file(self, tmp_path: Path):
+        mcp = FastMCP()
+
+        @mcp.tool
+        def file_tool(path: str) -> File:
+            return File(path)
+
+        # Create a test file
+        file_path = tmp_path / "test.bin"
+        file_path.write_bytes(b"test file data")
+
+        async with Client(mcp) as client:
+            result = await client.call_tool("file_tool", {"path": str(file_path)})
+            content = result[0]
+            assert isinstance(content, EmbeddedResource)
+            assert content.type == "resource"
+            resource = content.resource
+            assert resource.mimeType == "application/octet-stream"
+            # Verify base64 encoding
+            assert hasattr(resource, "blob")
+            blob_data = getattr(resource, "blob")
+            decoded = base64.b64decode(blob_data)
+            assert decoded == b"test file data"
+            # Verify URI points to the file
+            assert str(resource.uri) == file_path.resolve().as_uri()
+
     async def test_tool_mixed_content(self, tool_server: FastMCP):
         async with Client(tool_server) as client:
             result = await client.call_tool("mixed_content_tool", {})
-            assert len(result) == 2
+            assert len(result) == 3
             content1 = result[0]
             content2 = result[1]
+            content3 = result[2]
             assert isinstance(content1, TextContent)
             assert content1.text == "Hello"
             assert isinstance(content2, ImageContent)
-            assert content2.mimeType == "image/png"
+            assert content2.mimeType == "application/octet-stream"
             assert content2.data == "abc"
+            assert isinstance(content3, EmbeddedResource)
+            assert content3.type == "resource"
+            resource = content3.resource
+            assert resource.mimeType == "application/octet-stream"
+            assert hasattr(resource, "blob")
+            blob_data = getattr(resource, "blob")
+            decoded = base64.b64decode(blob_data)
+            assert decoded == b"abc"
 
     async def test_tool_mixed_list_with_image(
         self, tool_server: FastMCP, tmp_path: Path
@@ -372,6 +445,38 @@ class TestToolReturnTypes:
             assert isinstance(content3, TextContent)
             assert content3.text == "direct content"
 
+    async def test_tool_mixed_list_with_file(
+        self, tool_server: FastMCP, tmp_path: Path
+    ):
+        """Test that lists containing File objects and other types are handled
+        correctly. Note that the non-MCP content will be grouped together."""
+        # Create a test file
+        file_path = tmp_path / "test.bin"
+        file_path.write_bytes(b"test file data")
+
+        async with Client(tool_server) as client:
+            result = await client.call_tool(
+                "mixed_file_list_fn", {"file_path": str(file_path)}
+            )
+            assert len(result) == 3
+            # Check text conversion
+            content1 = result[0]
+            assert isinstance(content1, TextContent)
+            assert json.loads(content1.text) == ["text message", {"key": "value"}]
+            # Check file conversion
+            content2 = result[1]
+            assert isinstance(content2, EmbeddedResource)
+            assert content2.type == "resource"
+            resource = content2.resource
+            assert resource.mimeType == "application/octet-stream"
+            assert hasattr(resource, "blob")
+            blob_data = getattr(resource, "blob")
+            assert base64.b64decode(blob_data) == b"test file data"
+            # Check direct TextContent
+            content3 = result[2]
+            assert isinstance(content3, TextContent)
+            assert content3.text == "direct content"
+
 
 class TestToolParameters:
     async def test_parameter_descriptions_with_field_annotations(self):

tests/tools/test_tool.py

@@ -13,7 +13,7 @@ from fastmcp.client import Client
 from fastmcp.exceptions import ToolError
 from fastmcp.tools.tool import Tool, _convert_to_content
 from fastmcp.utilities.tests import temporary_settings
-from fastmcp.utilities.types import Audio, Image
+from fastmcp.utilities.types import Audio, File, Image
 
 
 class TestToolFromFunction:
@@ -114,6 +114,21 @@ class TestToolFromFunction:
         assert tool.parameters["properties"]["data"]["type"] == "string"
         assert isinstance(result[0], AudioContent)
 
+    async def test_tool_with_file_return(self):
+        def file_tool(data: bytes) -> File:
+            return File(data=data, format="octet-stream")
+
+        tool = Tool.from_function(file_tool)
+
+        result = await tool.run({"data": "test.bin"})
+        assert tool.parameters["properties"]["data"]["type"] == "string"
+        assert len(result) == 1
+        assert isinstance(result[0], EmbeddedResource)
+        assert result[0].type == "resource"
+        assert hasattr(result[0], "resource")
+        resource = result[0].resource
+        assert resource.mimeType == "application/octet-stream"
+
     def test_non_callable_fn(self):
         with pytest.raises(TypeError, match="not a callable object"):
             Tool.from_function(1)  # type: ignore
@@ -468,6 +483,38 @@ class TestConvertResultToContent:
         assert isinstance(result[0], AudioContent)
         assert result[0].data == "ZmFrZWF1ZGlvZGF0YQ=="
 
+    def test_file_object_result(self):
+        """Test that a File object is converted to EmbeddedResource with BlobResourceContents."""
+        file_obj = File(data=b"filedata", format="octet-stream")
+
+        result = _convert_to_content(file_obj)
+
+        assert isinstance(result, list)
+        assert len(result) == 1
+        assert isinstance(result[0], EmbeddedResource)
+        assert result[0].type == "resource"
+        assert hasattr(result[0], "resource")
+        resource = result[0].resource
+        assert resource.mimeType == "application/octet-stream"
+        # Check for blob attribute and its value
+        assert hasattr(resource, "blob")
+        assert getattr(resource, "blob") == "ZmlsZWRhdGE="  # base64 encoded "filedata"
+        # Convert URI to string for startswith check
+        assert str(resource.uri).startswith("file:///resource.octet-stream")
+
+    def test_file_object_text_result(self):
+        """Test that a File object with text data is converted to EmbeddedResource with TextResourceContents."""
+        file_obj = File(data=b"sometext", format="plain")
+        result = _convert_to_content(file_obj)
+        assert isinstance(result, list)
+        assert len(result) == 1
+        assert isinstance(result[0], EmbeddedResource)
+        assert result[0].type == "resource"
+        resource = result[0].resource
+        assert isinstance(resource, TextResourceContents)
+        assert resource.mimeType == "text/plain"
+        assert resource.text == "sometext"
+
     def test_basic_type_result(self):
         """Test that a basic type is converted to TextContent."""
         result = _convert_to_content(123)
@@ -574,6 +621,39 @@ class TestConvertResultToContent:
         audio_item = next(item for item in result if isinstance(item, AudioContent))
         assert audio_item.data == "ZmFrZWF1ZGlvZGF0YQ=="
 
+    def test_list_of_mixed_types_with_file(self):
+        """Test that a list of mixed types including File is converted correctly."""
+        content1 = TextContent(type="text", text="hello")
+        file_obj = File(data=b"filedata", format="octet-stream")
+        basic_data = {"a": 1}
+        result = _convert_to_content([content1, file_obj, basic_data])
+
+        assert isinstance(result, list)
+        assert len(result) == 3
+
+        text_content_count = sum(isinstance(item, TextContent) for item in result)
+        embedded_content_count = sum(
+            isinstance(item, EmbeddedResource) and item.type == "resource"
+            for item in result
+        )
+
+        assert text_content_count == 2
+        assert embedded_content_count == 1
+
+        text_item = next(item for item in result if isinstance(item, TextContent))
+        assert text_item.text == '{\n  "a": 1\n}'
+
+        embedded_item = next(
+            item
+            for item in result
+            if isinstance(item, EmbeddedResource) and item.type == "resource"
+        )
+        resource = embedded_item.resource
+        assert resource.mimeType == "application/octet-stream"
+        # Check for blob attribute and its value
+        assert hasattr(resource, "blob")
+        assert getattr(resource, "blob") == "ZmlsZWRhdGE="
+
     def test_empty_list(self):
         """Test that an empty list results in an empty list."""
         result = _convert_to_content([])

tests/utilities/test_types.py

@@ -3,9 +3,11 @@ from types import EllipsisType
 from typing import Annotated, Any
 
 import pytest
+from mcp.types import BlobResourceContents, TextResourceContents
 
 from fastmcp.utilities.types import (
     Audio,
+    File,
     Image,
     find_kwarg_by_type,
     is_class_member_of_type,
@@ -300,6 +302,118 @@ class TestAudio:
         assert content.data == base64.b64encode(test_data).decode()
 
 
+class TestFile:
+    def test_file_initialization_with_path(self):
+        """Test file initialization with a path."""
+        # Mock test - we're not actually going to read a file
+        file = File(path="test.txt")
+        assert file.path is not None
+        assert file.data is None
+        assert file._mime_type == "text/plain"
+
+    def test_file_initialization_with_data(self):
+        """Test initialization with data and format."""
+        test_data = b"test data"
+        file = File(data=test_data, format="octet-stream")
+        assert file.data == test_data
+        # The format parameter should set the MIME type
+        assert file._mime_type == "application/octet-stream"
+        assert file._name is None
+        assert file.annotations is None
+
+    def test_file_initialization_with_format(self):
+        """Test file initialization with a specific format."""
+        file = File(data=b"test", format="pdf")
+        assert file._mime_type == "application/pdf"
+
+    def test_file_initialization_with_name(self):
+        """Test file initialization with a custom name."""
+        file = File(data=b"test", name="custom")
+        assert file._name == "custom"
+
+    def test_missing_data_and_path_raises_error(self):
+        """Test that error is raised when neither path nor data is provided."""
+        with pytest.raises(ValueError, match="Either path or data must be provided"):
+            File()
+
+    def test_both_data_and_path_raises_error(self):
+        """Test that error is raised when both path and data are provided."""
+        with pytest.raises(
+            ValueError, match="Only one of path or data can be provided"
+        ):
+            File(path="test.txt", data=b"test")
+
+    def test_get_mime_type_from_path(self, tmp_path):
+        """Test MIME type detection from file extension."""
+        file_path = tmp_path / "test.txt"
+        file_path.write_text(
+            "test content"
+        )  # Need to write content for MIME type detection
+        file = File(path=file_path)
+        # The MIME type should be detected from the .txt extension
+        assert file._mime_type == "text/plain"
+
+    def test_to_resource_content_with_path(self, tmp_path):
+        """Test conversion to ResourceContent with path."""
+        file_path = tmp_path / "test.txt"
+        test_data = b"test file data"
+        file_path.write_bytes(test_data)
+
+        file = File(path=file_path)
+        resource = file.to_resource_content()
+
+        assert resource.type == "resource"
+        assert resource.resource.mimeType == "text/plain"
+        # Convert both to strings for comparison
+        assert str(resource.resource.uri) == file_path.resolve().as_uri()
+        if isinstance(resource.resource, BlobResourceContents):
+            assert resource.resource.blob == base64.b64encode(test_data).decode()
+
+    def test_to_resource_content_with_data(self):
+        """Test conversion to ResourceContent with data."""
+        test_data = b"test file data"
+        file = File(data=test_data, format="pdf")
+        resource = file.to_resource_content()
+
+        assert resource.type == "resource"
+        assert resource.resource.mimeType == "application/pdf"
+        # Convert URI to string for comparison
+        assert str(resource.resource.uri) == "file:///resource.pdf"
+        if isinstance(resource.resource, BlobResourceContents):
+            assert resource.resource.blob == base64.b64encode(test_data).decode()
+
+    def test_to_resource_content_with_text_data(self):
+        """Test conversion to ResourceContent with text data (TextResourceContents)."""
+        test_data = b"hello world"
+        file = File(data=test_data, format="plain")
+        resource = file.to_resource_content()
+        assert resource.type == "resource"
+        # Should be TextResourceContents for text/plain
+        assert isinstance(resource.resource, TextResourceContents)
+        assert resource.resource.mimeType == "text/plain"
+        assert resource.resource.text == "hello world"
+
+    def test_to_resource_content_error(self, monkeypatch):
+        """Test error case in to_resource_content."""
+        file = File(data=b"test")
+        monkeypatch.setattr(file, "path", None)
+        monkeypatch.setattr(file, "data", None)
+
+        with pytest.raises(ValueError, match="No resource data available"):
+            file.to_resource_content()
+
+    def test_to_resource_content_with_override_mime_type(self, tmp_path):
+        """Test conversion to ResourceContent with override MIME type."""
+        file_path = tmp_path / "test.txt"
+        test_data = b"test file data"
+        file_path.write_bytes(test_data)
+
+        file = File(path=file_path)
+        resource = file.to_resource_content(mime_type="application/custom")
+
+        assert resource.resource.mimeType == "application/custom"
+
+
 class TestFindKwargByType:
     def test_exact_type_match(self):
         """Test finding parameter with exact type match."""