Add file utility type and tests

src/fastmcp/tools/tool.py

@@ -19,6 +19,7 @@ from fastmcp.utilities.logging import get_logger
 from fastmcp.utilities.types import (
     Audio,
     Image,
+    File,
     MCPContent,
     find_kwarg_by_type,
     get_cached_typeadapter,
@@ -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

@@ -7,6 +7,7 @@ from functools import lru_cache
 from pathlib import Path
 from types import UnionType
 from typing import Annotated, TypeAlias, TypeVar, Union, get_args, get_origin
+import mimetypes
 
 from mcp.types import (
     Annotations,
@@ -14,6 +15,7 @@ from mcp.types import (
     EmbeddedResource,
     ImageContent,
     TextContent,
+    BlobResourceContents
 )
 from pydantic import BaseModel, ConfigDict, TypeAdapter
 
@@ -203,3 +205,68 @@ 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:
+            return f"application/{self._format.lower()}"
+
+        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:
+                data = base64.b64encode(f.read()).decode()
+                uri=self.path.resolve().as_uri()
+
+        elif self.data is not None:
+            data = base64.b64encode(self.data).decode()
+            uri=self.path or (self._name and f"file:///{self._name}.{self._mime_type.split('/')[1]}") or f"file:///resource.{self._mime_type.split('/')[1]}"
+
+        else:
+            raise ValueError("No resource data available")
+
+        resource = BlobResourceContents(
+                blob=data, 
+                mimeType=mime_type or self._mime_type,
+                uri=uri,
+            )
+
+        return EmbeddedResource(
+            type="resource",
+            resource=resource,
+            annotations=annotations or self.annotations,
+        )

tests/server/test_server_interactions.py

@@ -15,6 +15,7 @@ from mcp.types import (
     ImageContent,
     TextContent,
     TextResourceContents,
+    BlobResourceContents,
 )
 from pydantic import AnyUrl, Field
 
@@ -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,15 @@ 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="abc", mimeType="application/octet-stream", uri=AnyUrl("abc"))),
         ]
 
     @mcp.tool
@@ -77,6 +83,15 @@ 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"),
+        ]
+
     return mcp
 
 
@@ -88,7 +103,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()) == 10
 
     async def test_call_tool(self, tool_server: FastMCP):
         async with Client(tool_server) as client:
@@ -304,17 +319,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 +422,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:
@@ -113,6 +113,21 @@ class TestToolFromFunction:
         result = await tool.run({"data": "test.wav"})
         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"):
@@ -468,6 +483,25 @@ 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_basic_type_result(self):
         """Test that a basic type is converted to TextContent."""
         result = _convert_to_content(123)
@@ -574,6 +608,38 @@ 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

@@ -6,6 +6,7 @@ import pytest
 
 from fastmcp.utilities.types import (
     Audio,
+    File,
     Image,
     find_kwarg_by_type,
     is_class_member_of_type,
@@ -300,6 +301,103 @@ 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()
+        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"
+        assert resource.resource.blob == base64.b64encode(test_data).decode()
+
+    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."""