Added Audio to FastMCP (#1130)

This commit is contained in:
xavier 2025-08-22 18:45:38 +08:00 • committed by GitHub
commit f4b2957a20
No known key found for this signature in database
GPG key ID: B5690EEEBB952194

View file

@ -10,7 +10,7 @@ from starlette.routing import Mount, Route
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.server.session import ServerSession
from mcp.shared.exceptions import McpError
from mcp.shared.memory import (
@ -195,6 +195,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"),
@ -300,6 +304,60 @@ class TestServerTools:
# 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.parametrize(
"filename,expected_mime_type",
[
("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
],
)
@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)
# Create a test audio file with the specific extension
audio_path = tmp_path / filename
audio_path.write_bytes(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):
mcp = FastMCP()
@ -332,19 +390,24 @@ class TestServerTools:
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")
# TODO(Marcelo): It seems if we add the proper type hint, it generates an invalid JSON schema.
# We need to fix this.
def mixed_list_fn() -> list: # type: ignore
return [ # type: ignore
"text message",
Image(image_path),
Audio(audio_path),
{"key": "value"},
TextContent(type="text", text="direct content"),
]
@ -353,7 +416,7 @@ class TestServerTools:
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)
@ -363,14 +426,19 @@ class TestServerTools:
assert isinstance(content2, ImageContent)
assert content2.mimeType == "image/png"
assert base64.b64decode(content2.data) == b"test image data"
# Check dict conversion
# Check audio 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"
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