Support ImageContent and AudioContent in Message class

Message.content now accepts ImageContent and AudioContent in addition to
TextContent and EmbeddedResource, matching MCP's ContentBlock type. This
fixes ProxyPrompt.render() silently JSON-serializing image/audio content
instead of preserving it.

🤖 Generated with Claude Code

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Eric Robinson 2026-03-05 15:53:54 -06:00 committed by Jeremiah Lowin
commit 9319a2c645
3 changed files with 76 additions and 4 deletions

View file

@ -17,8 +17,10 @@ if TYPE_CHECKING:
import mcp.types
from mcp import GetPromptResult
from mcp.types import (
AudioContent,
EmbeddedResource,
Icon,
ImageContent,
PromptMessage,
TextContent,
)
@ -61,7 +63,7 @@ class Message(pydantic.BaseModel):
"""
role: Literal["user", "assistant"]
content: TextContent | EmbeddedResource
content: TextContent | ImageContent | AudioContent | EmbeddedResource
def __init__(
self,
@ -72,13 +74,18 @@ class Message(pydantic.BaseModel):
Args:
content: The message content. str passes through directly.
TextContent and EmbeddedResource pass through.
TextContent, ImageContent, AudioContent, and
EmbeddedResource pass through.
Other types (dict, list, BaseModel) are JSON-serialized.
role: The message role, either "user" or "assistant".
"""
# Handle already-wrapped content types
if isinstance(content, (TextContent, EmbeddedResource)):
normalized_content: TextContent | EmbeddedResource = content
if isinstance(
content, (TextContent, ImageContent, AudioContent, EmbeddedResource)
):
normalized_content: (
TextContent | ImageContent | AudioContent | EmbeddedResource
) = content
elif isinstance(content, str):
normalized_content = TextContent(type="text", text=content)
else:

View file

@ -493,6 +493,36 @@ class TestMessage:
assert isinstance(mcp_msg.content, TextContent)
assert mcp_msg.content.text == "Hello"
def test_message_passthrough_image_content(self):
"""Test Message passes through ImageContent without JSON serialization."""
from mcp.types import ImageContent
img = ImageContent(type="image", data="base64data", mimeType="image/png")
msg = Message(img, role="user")
assert isinstance(msg.content, ImageContent)
assert msg.content.data == "base64data"
assert msg.content.mimeType == "image/png"
def test_message_passthrough_audio_content(self):
"""Test Message passes through AudioContent without JSON serialization."""
from mcp.types import AudioContent
audio = AudioContent(type="audio", data="base64audio", mimeType="audio/wav")
msg = Message(audio, role="user")
assert isinstance(msg.content, AudioContent)
assert msg.content.data == "base64audio"
assert msg.content.mimeType == "audio/wav"
def test_message_image_content_to_mcp_prompt_message(self):
"""Test that ImageContent round-trips through to_mcp_prompt_message."""
from mcp.types import ImageContent
img = ImageContent(type="image", data="base64data", mimeType="image/png")
msg = Message(img, role="user")
mcp_msg = msg.to_mcp_prompt_message()
assert isinstance(mcp_msg.content, ImageContent)
assert mcp_msg.content.data == "base64data"
class TestPromptResult:
def test_promptresult_from_string(self):

View file

@ -130,6 +130,25 @@ def fastmcp_server():
def welcome(name: str) -> str:
return f"Welcome to FastMCP, {name}!"
@server.prompt
def image_prompt():
"""A prompt that returns an image."""
from fastmcp.prompts.prompt import Message, PromptResult
return PromptResult(
messages=[
Message("Here is an image:"),
Message(
content=mcp_types.ImageContent(
type="image",
data="iVBORw0KGgoAAAANSUhEUg==",
mimeType="image/png",
),
role="user",
),
]
)
return server
@ -656,6 +675,22 @@ class TestPrompts:
param_names = [arg.name for arg in welcome_prompt.arguments or []]
assert "extra" in param_names
async def test_proxy_prompt_preserves_image_content(
self, fastmcp_server: FastMCP, proxy_server: FastMCPProxy
):
"""Test that ProxyPrompt preserves ImageContent without lossy conversion."""
async with Client(fastmcp_server) as client:
result = await client.get_prompt("image_prompt")
async with Client(proxy_server) as client:
proxy_result = await client.get_prompt("image_prompt")
# The proxy result should match the original exactly
assert proxy_result == result
# Verify the image content is preserved as ImageContent, not JSON text
assert isinstance(proxy_result.messages[1].content, mcp_types.ImageContent)
assert proxy_result.messages[1].content.data == "iVBORw0KGgoAAAANSUhEUg=="
assert proxy_result.messages[1].content.mimeType == "image/png"
async def test_proxy_handles_multiple_concurrent_tasks_correctly(
proxy_server: FastMCPProxy,