MCPServer: content-block returns are unstructured, prompt messages take Image/Audio

docs/servers/structured-output.md

@@ -208,6 +208,10 @@ No `output_schema`, no wrapping, no validation. `structured_content` is `None` a
 
 The opposite, `structured_output=True`, turns the automatic detection into a requirement: a tool whose return type can't produce a schema raises at import time instead of falling back to text.
 
+## Content blocks and media
+
+Content blocks and media (`TextContent`, `EmbeddedResource`, `Image`, `Audio` and friends, on their own, as the items of a `list`, `tuple` or `Sequence`, or as the arms of a union) are opted out for you: they are for the model to read, so auto-detection derives no schema from them (**[Images, audio & icons](media.md)** covers `Image` and `Audio`). `structured_output=True` still forces one for the content-block classes.
+
 ## A class without type hints
 
 There is one way to end up unstructured without asking for it: return a class that has **no annotations on its body**.
@@ -240,6 +244,6 @@ There is one way to end up unstructured without asking for it: return a class th
 * Scalars, lists, tuples and unions are wrapped in `{"result": ...}`. Models, `TypedDict`s, dataclasses, annotated classes and `dict[str, ...]` are objects already and stay as they are.
 * Every result carries `content` (text, for the model) **and** `structured_content` (data, for the application).
 * What you return is validated against the schema. A mismatch is a tool error, not a corrupt result.
-* `structured_output=False` opts a tool out. A class without type hints opts out silently; watch for it.
+* `structured_output=False` opts a tool out. Content blocks, `Image` and `Audio` opt out by default; a class without type hints opts out silently, so watch for it.
 
 You now own everything a tool can say back. Next, the second primitive: **[Resources](resources.md)**.

src/mcp/client/client.py

@@ -946,5 +946,4 @@ async def list_tools(
     @deprecated("The roots capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
     async def send_roots_list_changed(self) -> None:
         """Send a notification that the roots list has changed."""
-        # TODO(Marcelo): Currently, there is no way for the server to handle this. We should add support.
         await self.session.send_roots_list_changed()  # pyright: ignore[reportDeprecated]

src/mcp/server/mcpserver/__init__.py

@@ -13,6 +13,7 @@
 )
 
 from .context import Context
+from .prompts.base import AssistantMessage, Message, UserMessage
 from .resolve import (
     AcceptedElicitation,
     CancelledElicitation,
@@ -32,6 +33,9 @@
     "Context",
     "Image",
     "Audio",
+    "Message",
+    "UserMessage",
+    "AssistantMessage",
     "Icon",
     "Resolve",
     "Elicit",

src/mcp/server/mcpserver/prompts/base.py

@@ -4,7 +4,7 @@
 
 import functools
 from collections.abc import Awaitable, Callable, Sequence
-from typing import TYPE_CHECKING, Any, Literal
+from typing import TYPE_CHECKING, Annotated, Any, Literal
 
 import anyio.to_thread
 import pydantic_core
@@ -13,6 +13,7 @@
 
 from mcp.server.mcpserver.utilities.context_injection import find_context_parameter, inject_context
 from mcp.server.mcpserver.utilities.func_metadata import func_metadata
+from mcp.server.mcpserver.utilities.types import Audio, Image
 from mcp.shared._callable_inspection import is_async_callable
 from mcp.shared.exceptions import MCPError
 
@@ -22,14 +23,22 @@
 
 
 class Message(BaseModel):
-    """Base class for all prompt messages."""
+    """Base class for all prompt messages.
+
+    `content` may be a plain string (wrapped in `TextContent`), an `Image` or `Audio`
+    helper (converted to `ImageContent` / `AudioContent`), or any ready-made content block.
+    """
 
     role: Literal["user", "assistant"]
     content: ContentBlock
 
-    def __init__(self, content: str | ContentBlock, **kwargs: Any):
+    def __init__(self, content: str | ContentBlock | Image | Audio, **kwargs: Any):
         if isinstance(content, str):
             content = TextContent(type="text", text=content)
+        elif isinstance(content, Image):
+            content = content.to_image_content()
+        elif isinstance(content, Audio):
+            content = content.to_audio_content()
         super().__init__(content=content, **kwargs)
 
 
@@ -38,7 +47,7 @@
 
     role: Literal["user", "assistant"] = "user"
 
-    def __init__(self, content: str | ContentBlock, **kwargs: Any):
+    def __init__(self, content: str | ContentBlock | Image | Audio, **kwargs: Any):
         super().__init__(content=content, **kwargs)
 
 
@@ -47,13 +56,18 @@
 
     role: Literal["user", "assistant"] = "assistant"
 
-    def __init__(self, content: str | ContentBlock, **kwargs: Any):
+    def __init__(self, content: str | ContentBlock | Image | Audio, **kwargs: Any):
         super().__init__(content=content, **kwargs)
 
 
-message_validator = TypeAdapter[UserMessage | AssistantMessage](UserMessage | AssistantMessage)
+# Both classes accept either role, so the first arm always matches: validate left to right rather than
+# trying both (which converted - and for path-backed Image/Audio, read - the content twice).
+message_validator: TypeAdapter[UserMessage | AssistantMessage] = TypeAdapter(
+    Annotated[UserMessage | AssistantMessage, Field(union_mode="left_to_right")]
+)
 
-SyncPromptResult = str | Message | dict[str, Any] | InputRequiredResult | Sequence[str | Message | dict[str, Any]]
+_PromptResultItem = str | ContentBlock | Image | Audio | Message | dict[str, Any]
+SyncPromptResult = _PromptResultItem | InputRequiredResult | Sequence[_PromptResultItem]
 PromptResult = SyncPromptResult | Awaitable[SyncPromptResult]
 
 
@@ -89,7 +103,7 @@
         """Create a Prompt from a function.
 
         The function can return:
-        - A string (converted to a message)
+        - A string, content block, `Image` or `Audio` (each becomes a user message)
         - A Message object
         - A dict (converted to a message)
         - A sequence of any of the above
@@ -105,10 +119,9 @@
         if context_kwarg is None:  # pragma: no branch
             context_kwarg = find_context_parameter(fn)
 
-        # Get schema from func_metadata, excluding context parameter
+        # Only the argument model is needed; a prompt has no output schema to derive
         func_arg_metadata = func_metadata(
-            fn,
-            skip_names=[context_kwarg] if context_kwarg is not None else [],
+            fn, skip_names=[context_kwarg] if context_kwarg is not None else [], structured_output=False
         )
         parameters = func_arg_metadata.arg_model.model_json_schema()
 
@@ -179,19 +192,15 @@
             # Convert result to messages
             messages: list[Message] = []
             for msg in result:  # type: ignore[reportUnknownVariableType]
-                try:
-                    if isinstance(msg, Message):
-                        messages.append(msg)
-                    elif isinstance(msg, dict):
-                        messages.append(message_validator.validate_python(msg))
-                    elif isinstance(msg, str):
-                        content = TextContent(type="text", text=msg)
-                        messages.append(UserMessage(content=content))
-                    else:  # pragma: no cover
-                        content = pydantic_core.to_json(msg, fallback=str, indent=2).decode()
-                        messages.append(Message(role="user", content=content))
-                except Exception:  # pragma: no cover
-                    raise ValueError(f"Could not convert prompt result to message: {msg}")
+                if isinstance(msg, Message):
+                    messages.append(msg)
+                elif isinstance(msg, dict):
+                    messages.append(message_validator.validate_python(msg))
+                elif isinstance(msg, str | ContentBlock | Image | Audio):  # bare content is one user message
+                    messages.append(UserMessage(msg))
+                else:  # pragma: no cover
+                    content = pydantic_core.to_json(msg, fallback=str, indent=2).decode()
+                    messages.append(Message(role="user", content=content))
 
             return messages
         except MCPError:

src/mcp/server/mcpserver/resources/templates.py

@@ -152,10 +152,9 @@ def from_function(
         if context_kwarg is None:  # pragma: no branch
             context_kwarg = find_context_parameter(fn)
 
-        # Get schema from func_metadata, excluding context parameter
+        # Only the argument model is needed; a resource has no output schema to derive
         func_arg_metadata = func_metadata(
-            fn,
-            skip_names=[context_kwarg] if context_kwarg is not None else [],
+            fn, skip_names=[context_kwarg] if context_kwarg is not None else [], structured_output=False
         )
         parameters = func_arg_metadata.arg_model.model_json_schema()
 

src/mcp/server/mcpserver/server.py

@@ -918,8 +918,8 @@ def prompt(
     ) -> Callable[[_CallableT], _CallableT]:
         """Decorator to register a prompt.
 
-        The function returns the prompt messages (a string, `Message`, dict,
-        or a sequence of these), or an `InputRequiredResult` to request
+        The function returns the prompt messages (a string, content block, `Image`/`Audio`,
+        `Message`, dict, or a sequence of these), or an `InputRequiredResult` to request
         client input first (the 2026-07-28 multi-round-trip flow — read
         `ctx.input_responses` on the retry).
 

src/mcp/server/mcpserver/utilities/func_metadata.py

@@ -33,6 +33,25 @@ def _is_input_required_type(obj: Any) -> bool:
     return isinstance(obj, type) and issubclass(obj, InputRequiredResult)
 
 
+_CONTENT_TYPES = (*get_args(ContentBlock), Image, Audio)
+# `_convert_to_content` unrolls list/tuple values; a `Sequence[...]` annotation is one of those at runtime.
+_CONTENT_SEQUENCE_ORIGINS = (list, tuple, Sequence)
+
+
+def _returns_content(annotation: Any) -> bool:
+    """Whether a return annotation declares content blocks or the `Image`/`Audio` helpers, bare or as
+    the items of a list/tuple or the arms of a union: the values `_convert_to_content` renders as blocks
+    rather than dumping as data. Keep the two in sync."""
+    origin = get_origin(annotation)
+    if origin is None:
+        return isinstance(annotation, type) and issubclass(annotation, _CONTENT_TYPES)
+    if origin is Annotated:
+        return _returns_content(get_args(annotation)[0])
+    if is_union_origin(origin) or origin in _CONTENT_SEQUENCE_ORIGINS:
+        return any(_returns_content(arg) for arg in get_args(annotation))
+    return False
+
+
 class StrictJsonSchema(GenerateJsonSchema):
     """A JSON schema generator that raises exceptions instead of emitting warnings.
 
@@ -222,6 +241,9 @@ def func_metadata(
             - TypedDict - converted to a Pydantic model with same fields
             - Dataclasses and other annotated classes - converted to Pydantic models
             - Generic types (list, dict, Union, etc.) - wrapped in a model with a 'result' field
+            - Content blocks (TextContent, EmbeddedResource, ...), Image and Audio, bare or inside a
+                list, tuple or union - unstructured when auto-detecting; structured_output=True bypasses
+                this rule (a content block then publishes its own schema; Image/Audio have none and raise)
 
     Returns:
         A FuncMetadata object containing:
@@ -345,6 +367,13 @@ def func_metadata(
     else:
         original_annotation = effective_annotation
 
+    if structured_output is None and _returns_content(return_type_expr):
+        # Content blocks and the Image/Audio helpers are what the model reads, not data for the
+        # application: a derived schema would advertise the block's own model as output_schema (and,
+        # unless the tool builds its own CallToolResult, echo every block into structured_content).
+        # structured_output=True still forces one.
+        return FuncMetadata(arg_model=arguments_model)
+
     output_model, output_schema, wrap_output = _try_create_model_and_schema(
         original_annotation, return_type_expr, func.__name__
     )
@@ -546,7 +575,7 @@ def _convert_to_content(result: Any) -> list[ContentBlock]:
     Note: This conversion logic comes from previous versions of MCPServer and is being
     retained for purposes of backwards compatibility. It produces different unstructured
     output than the lowlevel server tool call handler, which just serializes structured
-    content verbatim.
+    content verbatim. `_returns_content` is the annotation-level mirror of these branches.
     """
     if result is None:  # pragma: no cover
         return []

tests/docs_src/test_structured_output.py

@@ -2,7 +2,7 @@
 
 import pytest
 from inline_snapshot import snapshot
-from mcp_types import TextContent
+from mcp_types import EmbeddedResource, ImageContent, TextContent, TextResourceContents
 
 from docs_src.structured_output import (
     tutorial001,
@@ -17,6 +17,7 @@
 )
 from mcp import Client
 from mcp.server import MCPServer
+from mcp.server.mcpserver import Image
 from mcp.server.mcpserver.exceptions import InvalidSignature
 
 # See test_index.py for why this is a per-module mark and not a conftest hook.
@@ -173,6 +174,36 @@ async def test_structured_output_false_opts_out() -> None:
         ]
 
 
+async def test_content_blocks_and_media_are_opted_out_of_structured_output() -> None:
+    """The "Content blocks and media" section: a content-block or `Image`/`Audio` return annotation, bare or
+    as list items, derives no output schema and no structured content; the blocks are the result."""
+    mcp = MCPServer("Reports")
+    document = EmbeddedResource(
+        type="resource", resource=TextResourceContents(uri="report://q3", mime_type="text/markdown", text="# Q3")
+    )
+
+    @mcp.tool()
+    def report() -> EmbeddedResource:
+        return document
+
+    @mcp.tool()
+    def chart() -> list[str | Image]:
+        return ["Sales by region:", Image(data=b"png", format="png")]
+
+    async with Client(mcp) as client:
+        tools = {tool.name: tool for tool in (await client.list_tools()).tools}
+        assert tools["report"].output_schema is None
+        assert tools["chart"].output_schema is None
+        report_result = await client.call_tool("report", {})
+        assert (report_result.content, report_result.structured_content) == ([document], None)
+        chart_result = await client.call_tool("chart", {})
+        assert chart_result.structured_content is None
+        assert chart_result.content == [
+            TextContent(type="text", text="Sales by region:"),
+            ImageContent(type="image", data="cG5n", mime_type="image/png"),
+        ]
+
+
 async def test_class_without_type_hints_is_silently_unstructured() -> None:
     """tutorial009: a class with no annotations on its body gets no schema, and the model gets a `repr`."""
     async with Client(tutorial009.mcp) as client: