diff --git a/docs/servers/tools.mdx b/docs/servers/tools.mdx index 13a88f688..300a6c7dd 100644 --- a/docs/servers/tools.mdx +++ b/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). diff --git a/examples/get_file.py b/examples/get_file.py new file mode 100644 index 000000000..0b00ac47d --- /dev/null +++ b/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") diff --git a/src/fastmcp/tools/tool.py b/src/fastmcp/tools/tool.py index 14fe7f8b6..fd40f0d2a 100644 --- a/src/fastmcp/tools/tool.py +++ b/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) diff --git a/src/fastmcp/utilities/types.py b/src/fastmcp/utilities/types.py index 178a020f0..2397f73bf 100644 --- a/src/fastmcp/utilities/types.py +++ b/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, + ) diff --git a/tests/server/test_server_interactions.py b/tests/server/test_server_interactions.py index 75290edf1..6198a9f70 100644 --- a/tests/server/test_server_interactions.py +++ b/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): diff --git a/tests/tools/test_tool.py b/tests/tools/test_tool.py index 3a3388d3e..54ac2061e 100644 --- a/tests/tools/test_tool.py +++ b/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([]) diff --git a/tests/utilities/test_types.py b/tests/utilities/test_types.py index 8976c422d..1f8338318 100644 --- a/tests/utilities/test_types.py +++ b/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."""