From 992de1025536c4a87776e47cfe4f522db9f24b04 Mon Sep 17 00:00:00 2001 From: dragonier23 <84836280+dragonier23@users.noreply.github.com> Date: Sat, 12 Jul 2025 15:42:49 +0800 Subject: [PATCH 1/5] feat: Added Audio to FastMCP --- src/mcp/server/fastmcp/__init__.py | 4 +- .../server/fastmcp/utilities/func_metadata.py | 5 +- src/mcp/server/fastmcp/utilities/types.py | 51 +++++++++++++- tests/server/fastmcp/test_server.py | 68 ++++++++++++++++++- 4 files changed, 123 insertions(+), 5 deletions(-) diff --git a/src/mcp/server/fastmcp/__init__.py b/src/mcp/server/fastmcp/__init__.py index 84b052078..f8f9c1c4c 100644 --- a/src/mcp/server/fastmcp/__init__.py +++ b/src/mcp/server/fastmcp/__init__.py @@ -3,7 +3,7 @@ from importlib.metadata import version from .server import Context, FastMCP -from .utilities.types import Image +from .utilities.types import Audio, Image __version__ = version("mcp") -__all__ = ["FastMCP", "Context", "Image"] +__all__ = ["FastMCP", "Context", "Image", "Audio"] diff --git a/src/mcp/server/fastmcp/utilities/func_metadata.py b/src/mcp/server/fastmcp/utilities/func_metadata.py index 3c9506fba..f7743dd47 100644 --- a/src/mcp/server/fastmcp/utilities/func_metadata.py +++ b/src/mcp/server/fastmcp/utilities/func_metadata.py @@ -21,7 +21,7 @@ from mcp.server.fastmcp.exceptions import InvalidSignature from mcp.server.fastmcp.utilities.logging import get_logger -from mcp.server.fastmcp.utilities.types import Image +from mcp.server.fastmcp.utilities.types import Audio, Image from mcp.types import ContentBlock, TextContent logger = get_logger(__name__) @@ -480,6 +480,9 @@ def _convert_to_content( if isinstance(result, Image): return [result.to_image_content()] + if isinstance(result, Audio): + return [result.to_audio_content()] + if isinstance(result, list | tuple): return list( chain.from_iterable( diff --git a/src/mcp/server/fastmcp/utilities/types.py b/src/mcp/server/fastmcp/utilities/types.py index ccaa3d69a..da80b9996 100644 --- a/src/mcp/server/fastmcp/utilities/types.py +++ b/src/mcp/server/fastmcp/utilities/types.py @@ -3,7 +3,7 @@ import base64 from pathlib import Path -from mcp.types import ImageContent +from mcp.types import AudioContent, ImageContent class Image: @@ -52,3 +52,52 @@ def to_image_content(self) -> ImageContent: raise ValueError("No image data available") return ImageContent(type="image", data=data, mimeType=self._mime_type) + + +class Audio: + """Helper class for returning audio from tools.""" + + def __init__( + self, + path: str | Path | None = None, + data: bytes | None = None, + format: str | 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() + + def _get_mime_type(self) -> str: + """Get MIME type from format or guess from file extension.""" + if self._format: + return f"audio/{self._format.lower()}" + + if self.path: + suffix = self.path.suffix.lower() + return { + ".wav": "audio/wav", + ".mp3": "audio/mpeg", + ".ogg": "audio/ogg", + ".flac": "audio/flac", + ".aac": "audio/aac", + ".m4a": "audio/mp4", + }.get(suffix, "application/octet-stream") + return "audio/wav" # default for raw binary data + + def to_audio_content(self) -> AudioContent: + """Convert to MCP AudioContent.""" + if self.path: + with open(self.path, "rb") as f: + data = base64.b64encode(f.read()).decode() + elif self.data is not None: + data = base64.b64encode(self.data).decode() + else: + raise ValueError("No audio data available") + + return AudioContent(type="audio", data=data, mimeType=self._mime_type) diff --git a/tests/server/fastmcp/test_server.py b/tests/server/fastmcp/test_server.py index a9e0d182a..ae00077e0 100644 --- a/tests/server/fastmcp/test_server.py +++ b/tests/server/fastmcp/test_server.py @@ -10,7 +10,7 @@ from mcp.server.fastmcp import Context, FastMCP from mcp.server.fastmcp.prompts.base import Message, UserMessage from mcp.server.fastmcp.resources import FileResource, FunctionResource -from mcp.server.fastmcp.utilities.types import Image +from mcp.server.fastmcp.utilities.types import Audio, Image from mcp.shared.exceptions import McpError from mcp.shared.memory import ( create_connected_server_and_client_session as client_session, @@ -194,6 +194,10 @@ def image_tool_fn(path: str) -> Image: return Image(path) +def audio_tool_fn(path: str) -> Audio: + return Audio(path) + + def mixed_content_tool_fn() -> list[ContentBlock]: return [ TextContent(type="text", text="Hello"), @@ -299,6 +303,27 @@ async def test_tool_image_helper(self, tmp_path: Path): # Check structured content - Image return type should NOT have structured output assert result.structuredContent is None + @pytest.mark.anyio + async def test_tool_audio_helper(self, tmp_path: Path): + # Create a test audio + audio_path = tmp_path / "test.wav" + audio_path.write_bytes(b"fake wav data") + + mcp = FastMCP() + mcp.add_tool(audio_tool_fn) + async with client_session(mcp._mcp_server) as client: + result = await client.call_tool("audio_tool_fn", {"path": str(audio_path)}) + assert len(result.content) == 1 + content = result.content[0] + assert isinstance(content, AudioContent) + assert content.type == "audio" + assert content.mimeType == "audio/wav" + # Verify base64 encoding + decoded = base64.b64decode(content.data) + assert decoded == b"fake wav data" + # Check structured content - Image return type should NOT have structured output + assert result.structuredContent is None + @pytest.mark.anyio async def test_tool_mixed_content(self): mcp = FastMCP() @@ -371,6 +396,47 @@ def mixed_list_fn() -> list: # Check structured content - untyped list with Image objects should NOT have structured output assert result.structuredContent is None + @pytest.mark.anyio + async def test_tool_mixed_list_with_audio(self, tmp_path: Path): + """Test that lists containing Audio objects and other types are handled + correctly""" + # Create a test audio + audio_path = tmp_path / "test.wav" + audio_path.write_bytes(b"test audio data") + + def mixed_list_fn() -> list: + return [ + "text message", + Audio(audio_path), + {"key": "value"}, + TextContent(type="text", text="direct content"), + ] + + mcp = FastMCP() + mcp.add_tool(mixed_list_fn) + async with client_session(mcp._mcp_server) as client: + result = await client.call_tool("mixed_list_fn", {}) + assert len(result.content) == 4 + # Check text conversion + content1 = result.content[0] + assert isinstance(content1, TextContent) + assert content1.text == "text message" + # Check audio conversion + content2 = result.content[1] + assert isinstance(content2, AudioContent) + assert content2.mimeType == "audio/wav" + assert base64.b64decode(content2.data) == b"test audio data" + # Check dict conversion + content3 = result.content[2] + assert isinstance(content3, TextContent) + assert '"key": "value"' in content3.text + # Check direct TextContent + content4 = result.content[3] + assert isinstance(content4, TextContent) + assert content4.text == "direct content" + # Check structured content - untyped list with Audio objects should NOT have structured output + assert result.structuredContent is None + @pytest.mark.anyio async def test_tool_structured_output_basemodel(self): """Test tool with structured output returning BaseModel""" From 40294d54d3cddf0b7748a69df9b54d0c95230f97 Mon Sep 17 00:00:00 2001 From: dragonier23 <84836280+dragonier23@users.noreply.github.com> Date: Sat, 9 Aug 2025 22:13:24 +0800 Subject: [PATCH 2/5] fix: additional test for audio mimetype detection --- tests/server/fastmcp/test_server.py | 33 +++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/tests/server/fastmcp/test_server.py b/tests/server/fastmcp/test_server.py index ae00077e0..fdb4b91e4 100644 --- a/tests/server/fastmcp/test_server.py +++ b/tests/server/fastmcp/test_server.py @@ -324,6 +324,39 @@ async def test_tool_audio_helper(self, tmp_path: Path): # Check structured content - Image return type should NOT have structured output assert result.structuredContent is None + @pytest.mark.anyio + async def test_tool_audio_suffix_detection(self, tmp_path: Path): + # Test different audio file extensions + test_cases = [ + ("test.wav", "audio/wav"), + ("test.mp3", "audio/mpeg"), + ("test.ogg", "audio/ogg"), + ("test.flac", "audio/flac"), + ("test.aac", "audio/aac"), + ("test.m4a", "audio/mp4"), + ("test.unknown", "application/octet-stream"), # Unknown extension fallback + ] + + mcp = FastMCP() + mcp.add_tool(audio_tool_fn) + async with client_session(mcp._mcp_server) as client: + for filename, expected_mime_type in test_cases: + # Create a test audio file with the specific extension + audio_path = tmp_path / filename + audio_path.write_bytes(b"fake audio data") + + result = await client.call_tool("audio_tool_fn", {"path": str(audio_path)}) + assert len(result.content) == 1 + content = result.content[0] + assert isinstance(content, AudioContent) + assert content.type == "audio" + assert content.mimeType == expected_mime_type, ( + f"Expected {expected_mime_type} for {filename}, got {content.mimeType}" + ) + # Verify base64 encoding + decoded = base64.b64decode(content.data) + assert decoded == b"fake audio data" + @pytest.mark.anyio async def test_tool_mixed_content(self): mcp = FastMCP() From b9054600aa4df53fc547d92e46f6fa17d172b118 Mon Sep 17 00:00:00 2001 From: dragonier23 <84836280+dragonier23@users.noreply.github.com> Date: Sat, 16 Aug 2025 21:26:25 +0800 Subject: [PATCH 3/5] fix: tests for audio, init --- src/mcp/server/fastmcp/utilities/types.py | 8 +- tests/server/fastmcp/test_server.py | 107 ++++++++-------------- 2 files changed, 41 insertions(+), 74 deletions(-) diff --git a/src/mcp/server/fastmcp/utilities/types.py b/src/mcp/server/fastmcp/utilities/types.py index da80b9996..475d4ddd1 100644 --- a/src/mcp/server/fastmcp/utilities/types.py +++ b/src/mcp/server/fastmcp/utilities/types.py @@ -63,11 +63,9 @@ def __init__( data: bytes | None = None, format: str | 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") - + if not bool(path) ^ bool(data): + raise ValueError("Either path or data can be provided") + self.path = Path(path) if path else None self.data = data self._format = format diff --git a/tests/server/fastmcp/test_server.py b/tests/server/fastmcp/test_server.py index fdb4b91e4..a35477e34 100644 --- a/tests/server/fastmcp/test_server.py +++ b/tests/server/fastmcp/test_server.py @@ -324,10 +324,9 @@ async def test_tool_audio_helper(self, tmp_path: Path): # Check structured content - Image return type should NOT have structured output assert result.structuredContent is None - @pytest.mark.anyio - async def test_tool_audio_suffix_detection(self, tmp_path: Path): - # Test different audio file extensions - test_cases = [ + @pytest.mark.parametrize( + "filename,expected_mime_type", + [ ("test.wav", "audio/wav"), ("test.mp3", "audio/mpeg"), ("test.ogg", "audio/ogg"), @@ -335,27 +334,28 @@ async def test_tool_audio_suffix_detection(self, tmp_path: Path): ("test.aac", "audio/aac"), ("test.m4a", "audio/mp4"), ("test.unknown", "application/octet-stream"), # Unknown extension fallback - ] - + ], + ) + @pytest.mark.anyio + async def test_tool_audio_suffix_detection(self, tmp_path: Path, filename: str, expected_mime_type: str): + """Test that Audio helper correctly detects MIME types from file suffixes""" mcp = FastMCP() mcp.add_tool(audio_tool_fn) - async with client_session(mcp._mcp_server) as client: - for filename, expected_mime_type in test_cases: - # Create a test audio file with the specific extension - audio_path = tmp_path / filename - audio_path.write_bytes(b"fake audio data") + + # Create a test audio file with the specific extension + audio_path = tmp_path / filename + audio_path.write_bytes(b"fake audio data") - result = await client.call_tool("audio_tool_fn", {"path": str(audio_path)}) - assert len(result.content) == 1 - content = result.content[0] - assert isinstance(content, AudioContent) - assert content.type == "audio" - assert content.mimeType == expected_mime_type, ( - f"Expected {expected_mime_type} for {filename}, got {content.mimeType}" - ) - # Verify base64 encoding - decoded = base64.b64decode(content.data) - assert decoded == b"fake audio data" + async with client_session(mcp._mcp_server) as client: + result = await client.call_tool("audio_tool_fn", {"path": str(audio_path)}) + assert len(result.content) == 1 + content = result.content[0] + assert isinstance(content, AudioContent) + assert content.type == "audio" + assert content.mimeType == expected_mime_type + # Verify base64 encoding + decoded = base64.b64decode(content.data) + assert decoded == b"fake audio data" @pytest.mark.anyio async def test_tool_mixed_content(self): @@ -389,17 +389,22 @@ async def test_tool_mixed_content(self): assert structured_result[i][key] == value @pytest.mark.anyio - async def test_tool_mixed_list_with_image(self, tmp_path: Path): + async def test_tool_mixed_list_with_audio_and_image(self, tmp_path: Path): """Test that lists containing Image objects and other types are handled correctly""" # Create a test image image_path = tmp_path / "test.png" image_path.write_bytes(b"test image data") + # Create a test audio + audio_path = tmp_path / "test.wav" + audio_path.write_bytes(b"test audio data") + def mixed_list_fn() -> list: return [ "text message", Image(image_path), + Audio(audio_path), {"key": "value"}, TextContent(type="text", text="direct content"), ] @@ -418,56 +423,20 @@ def mixed_list_fn() -> list: assert isinstance(content2, ImageContent) assert content2.mimeType == "image/png" assert base64.b64decode(content2.data) == b"test image data" - # Check dict conversion - content3 = result.content[2] - assert isinstance(content3, TextContent) - assert '"key": "value"' in content3.text - # Check direct TextContent - content4 = result.content[3] - assert isinstance(content4, TextContent) - assert content4.text == "direct content" - # Check structured content - untyped list with Image objects should NOT have structured output - assert result.structuredContent is None - - @pytest.mark.anyio - async def test_tool_mixed_list_with_audio(self, tmp_path: Path): - """Test that lists containing Audio objects and other types are handled - correctly""" - # Create a test audio - audio_path = tmp_path / "test.wav" - audio_path.write_bytes(b"test audio data") - - def mixed_list_fn() -> list: - return [ - "text message", - Audio(audio_path), - {"key": "value"}, - TextContent(type="text", text="direct content"), - ] - - mcp = FastMCP() - mcp.add_tool(mixed_list_fn) - async with client_session(mcp._mcp_server) as client: - result = await client.call_tool("mixed_list_fn", {}) - assert len(result.content) == 4 - # Check text conversion - content1 = result.content[0] - assert isinstance(content1, TextContent) - assert content1.text == "text message" # Check audio conversion - content2 = result.content[1] - assert isinstance(content2, AudioContent) - assert content2.mimeType == "audio/wav" - assert base64.b64decode(content2.data) == b"test audio data" - # Check dict conversion content3 = result.content[2] - assert isinstance(content3, TextContent) - assert '"key": "value"' in content3.text - # Check direct TextContent + assert isinstance(content3, AudioContent) + assert content3.mimeType == "audio/wav" + assert base64.b64decode(content3.data) == b"test audio data" + # Check dict conversion content4 = result.content[3] assert isinstance(content4, TextContent) - assert content4.text == "direct content" - # Check structured content - untyped list with Audio objects should NOT have structured output + assert '"key": "value"' in content4.text + # Check direct TextContent + content5 = result.content[4] + assert isinstance(content5, TextContent) + assert content5.text == "direct content" + # Check structured content - untyped list with Image objects should NOT have structured output assert result.structuredContent is None @pytest.mark.anyio From 494f58a8fd27b5dae3b214caf2262d79ef0de95a Mon Sep 17 00:00:00 2001 From: dragonier23 <84836280+dragonier23@users.noreply.github.com> Date: Sat, 16 Aug 2025 21:33:23 +0800 Subject: [PATCH 4/5] fix: test --- tests/server/fastmcp/test_server.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/server/fastmcp/test_server.py b/tests/server/fastmcp/test_server.py index 0449ffa61..a8bc195ff 100644 --- a/tests/server/fastmcp/test_server.py +++ b/tests/server/fastmcp/test_server.py @@ -417,7 +417,7 @@ def mixed_list_fn() -> list: # type: ignore mcp.add_tool(mixed_list_fn) # type: ignore async with client_session(mcp._mcp_server) as client: result = await client.call_tool("mixed_list_fn", {}) - assert len(result.content) == 4 + assert len(result.content) == 5 # Check text conversion content1 = result.content[0] assert isinstance(content1, TextContent) From 1a043a4d26da062ce0d542650f132a95402157ec Mon Sep 17 00:00:00 2001 From: dragonier23 <84836280+dragonier23@users.noreply.github.com> Date: Sat, 16 Aug 2025 21:36:03 +0800 Subject: [PATCH 5/5] fix: linting problems --- src/mcp/server/fastmcp/utilities/types.py | 4 ++-- tests/server/fastmcp/test_server.py | 7 +++---- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/src/mcp/server/fastmcp/utilities/types.py b/src/mcp/server/fastmcp/utilities/types.py index 475d4ddd1..1be6f8274 100644 --- a/src/mcp/server/fastmcp/utilities/types.py +++ b/src/mcp/server/fastmcp/utilities/types.py @@ -63,9 +63,9 @@ def __init__( data: bytes | None = None, format: str | None = None, ): - if not bool(path) ^ bool(data): + if not bool(path) ^ bool(data): raise ValueError("Either path or data can be provided") - + self.path = Path(path) if path else None self.data = data self._format = format diff --git a/tests/server/fastmcp/test_server.py b/tests/server/fastmcp/test_server.py index a8bc195ff..3f921b588 100644 --- a/tests/server/fastmcp/test_server.py +++ b/tests/server/fastmcp/test_server.py @@ -10,7 +10,7 @@ from mcp.server.fastmcp import Context, FastMCP from mcp.server.fastmcp.prompts.base import Message, UserMessage from mcp.server.fastmcp.resources import FileResource, FunctionResource -from mcp.server.fastmcp.utilities.types import Image, Audio +from mcp.server.fastmcp.utilities.types import Audio, Image from mcp.server.session import ServerSession from mcp.shared.exceptions import McpError from mcp.shared.memory import ( @@ -342,7 +342,7 @@ async def test_tool_audio_suffix_detection(self, tmp_path: Path, filename: str, """Test that Audio helper correctly detects MIME types from file suffixes""" mcp = FastMCP() mcp.add_tool(audio_tool_fn) - + # Create a test audio file with the specific extension audio_path = tmp_path / filename audio_path.write_bytes(b"fake audio data") @@ -397,7 +397,7 @@ async def test_tool_mixed_list_with_audio_and_image(self, tmp_path: Path): image_path = tmp_path / "test.png" image_path.write_bytes(b"test image data") - # Create a test audio + # Create a test audio audio_path = tmp_path / "test.wav" audio_path.write_bytes(b"test audio data") @@ -405,7 +405,6 @@ async def test_tool_mixed_list_with_audio_and_image(self, tmp_path: Path): # We need to fix this. def mixed_list_fn() -> list: # type: ignore return [ # type: ignore - "text message", Image(image_path), Audio(audio_path),