modelcontextprotocol
diff --git a/README.md b/README.md
@@ -383,6 +383,61 @@ causes the tool to be classified as structured _and this is undesirable_,
 the  classification can be suppressed by passing `structured_output=False`
 to the `@tool` decorator.
 
+##### Advanced: Direct CallToolResult
+
+For full control over tool responses including the `_meta` field (for passing data to client applications without exposing it to the model), you can return `CallToolResult` directly:
+
+<!-- snippet-source examples/snippets/servers/direct_call_tool_result.py -->
+```python
+"""Example showing direct CallToolResult return for advanced control."""
+
+from typing import Annotated
+
+from pydantic import BaseModel
+
+from mcp.server.fastmcp import FastMCP
+from mcp.types import CallToolResult, TextContent
+
+mcp = FastMCP("CallToolResult Example")
+
+
+class ValidationModel(BaseModel):
+    """Model for validating structured output."""
+
+    status: str
+    data: dict[str, int]
+
+
+@mcp.tool()
+def advanced_tool() -> CallToolResult:
+    """Return CallToolResult directly for full control including _meta field."""
+    return CallToolResult(
+        content=[TextContent(type="text", text="Response visible to the model")],
+        _meta={"hidden": "data for client applications only"},
+    )
+
+
+@mcp.tool()
+def validated_tool() -> Annotated[CallToolResult, ValidationModel]:
+    """Return CallToolResult with structured output validation."""
+    return CallToolResult(
+        content=[TextContent(type="text", text="Validated response")],
+        structuredContent={"status": "success", "data": {"result": 42}},
+        _meta={"internal": "metadata"},
+    )
+
+
+@mcp.tool()
+def empty_result_tool() -> CallToolResult:
+    """For empty results, return CallToolResult with empty content."""
+    return CallToolResult(content=[])
+```
+
+_Full example: [examples/snippets/servers/direct_call_tool_result.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/direct_call_tool_result.py)_
+<!-- /snippet-source -->
+
+**Important:** `CallToolResult` must always be returned (no `Optional` or `Union`). For empty results, use `CallToolResult(content=[])`. For optional simple types, use `str | None` without `CallToolResult`.
+
 <!-- snippet-source examples/snippets/servers/structured_output.py -->
 ```python
 """Example showing structured output with tools."""
@@ -1769,14 +1824,93 @@ if __name__ == "__main__":
 _Full example: [examples/snippets/servers/lowlevel/structured_output.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/lowlevel/structured_output.py)_
 <!-- /snippet-source -->
 
-Tools can return data in three ways:
+Tools can return data in four ways:
 
 1. **Content only**: Return a list of content blocks (default behavior before spec revision 2025-06-18)
 2. **Structured data only**: Return a dictionary that will be serialized to JSON (Introduced in spec revision 2025-06-18)
 3. **Both**: Return a tuple of (content, structured_data) preferred option to use for backwards compatibility
+4. **Direct CallToolResult**: Return `CallToolResult` directly for full control (including `_meta` field)
 
 When an `outputSchema` is defined, the server automatically validates the structured output against the schema. This ensures type safety and helps catch errors early.
 
+##### Returning CallToolResult Directly
+
+For full control over the response including the `_meta` field (for passing data to client applications without exposing it to the model), return `CallToolResult` directly:
+
+<!-- snippet-source examples/snippets/servers/lowlevel/direct_call_tool_result.py -->
+```python
+"""
+Run from the repository root:
+    uv run examples/snippets/servers/lowlevel/direct_call_tool_result.py
+"""
+
+import asyncio
+from typing import Any
+
+import mcp.server.stdio
+import mcp.types as types
+from mcp.server.lowlevel import NotificationOptions, Server
+from mcp.server.models import InitializationOptions
+
+server = Server("example-server")
+
+
+@server.list_tools()
+async def list_tools() -> list[types.Tool]:
+    """List available tools."""
+    return [
+        types.Tool(
+            name="advanced_tool",
+            description="Tool with full control including _meta field",
+            inputSchema={
+                "type": "object",
+                "properties": {"message": {"type": "string"}},
+                "required": ["message"],
+            },
+        )
+    ]
+
+
+@server.call_tool()
+async def handle_call_tool(name: str, arguments: dict[str, Any]) -> types.CallToolResult:
+    """Handle tool calls by returning CallToolResult directly."""
+    if name == "advanced_tool":
+        message = str(arguments.get("message", ""))
+        return types.CallToolResult(
+            content=[types.TextContent(type="text", text=f"Processed: {message}")],
+            structuredContent={"result": "success", "message": message},
+            _meta={"hidden": "data for client applications only"},
+        )
+
+    raise ValueError(f"Unknown tool: {name}")
+
+
+async def run():
+    """Run the server."""
+    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
+        await server.run(
+            read_stream,
+            write_stream,
+            InitializationOptions(
+                server_name="example",
+                server_version="0.1.0",
+                capabilities=server.get_capabilities(
+                    notification_options=NotificationOptions(),
+                    experimental_capabilities={},
+                ),
+            ),
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(run())
+```
+
+_Full example: [examples/snippets/servers/lowlevel/direct_call_tool_result.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/lowlevel/direct_call_tool_result.py)_
+<!-- /snippet-source -->
+
+**Note:** When returning `CallToolResult`, you bypass the automatic content/structured conversion. You must construct the complete response yourself.
+
 ### Pagination (Advanced)
 
 For servers that need to handle large datasets, the low-level server provides paginated versions of list operations. This is an optional optimization - most servers won't need pagination unless they're dealing with hundreds or thousands of items.

diff --git a/examples/fastmcp/direct_call_tool_result_return.py b/examples/fastmcp/direct_call_tool_result_return.py
@@ -0,0 +1,24 @@
+"""
+FastMCP Echo Server with direct CallToolResult return
+"""
+
+from typing import Annotated
+
+from pydantic import BaseModel
+
+from mcp.server.fastmcp import FastMCP
+from mcp.types import CallToolResult, TextContent
+
+mcp = FastMCP("Echo Server")
+
+
+class EchoResponse(BaseModel):
+    text: str
+
+
+@mcp.tool()
+def echo(text: str) -> Annotated[CallToolResult, EchoResponse]:
+    """Echo the input text with structure and metadata"""
+    return CallToolResult(
+        content=[TextContent(type="text", text=text)], structuredContent={"text": text}, _meta={"some": "metadata"}
+    )
diff --git a/examples/snippets/servers/direct_call_tool_result.py b/examples/snippets/servers/direct_call_tool_result.py
@@ -0,0 +1,42 @@
+"""Example showing direct CallToolResult return for advanced control."""
+
+from typing import Annotated
+
+from pydantic import BaseModel
+
+from mcp.server.fastmcp import FastMCP
+from mcp.types import CallToolResult, TextContent
+
+mcp = FastMCP("CallToolResult Example")
+
+
+class ValidationModel(BaseModel):
+    """Model for validating structured output."""
+
+    status: str
+    data: dict[str, int]
+
+
+@mcp.tool()
+def advanced_tool() -> CallToolResult:
+    """Return CallToolResult directly for full control including _meta field."""
+    return CallToolResult(
+        content=[TextContent(type="text", text="Response visible to the model")],
+        _meta={"hidden": "data for client applications only"},
+    )
+
+
+@mcp.tool()
+def validated_tool() -> Annotated[CallToolResult, ValidationModel]:
+    """Return CallToolResult with structured output validation."""
+    return CallToolResult(
+        content=[TextContent(type="text", text="Validated response")],
+        structuredContent={"status": "success", "data": {"result": 42}},
+        _meta={"internal": "metadata"},
+    )
+
+
+@mcp.tool()
+def empty_result_tool() -> CallToolResult:
+    """For empty results, return CallToolResult with empty content."""
+    return CallToolResult(content=[])
diff --git a/examples/snippets/servers/lowlevel/direct_call_tool_result.py b/examples/snippets/servers/lowlevel/direct_call_tool_result.py
@@ -0,0 +1,65 @@
+"""
+Run from the repository root:
+    uv run examples/snippets/servers/lowlevel/direct_call_tool_result.py
+"""
+
+import asyncio
+from typing import Any
+
+import mcp.server.stdio
+import mcp.types as types
+from mcp.server.lowlevel import NotificationOptions, Server
+from mcp.server.models import InitializationOptions
+
+server = Server("example-server")
+
+
+@server.list_tools()
+async def list_tools() -> list[types.Tool]:
+    """List available tools."""
+    return [
+        types.Tool(
+            name="advanced_tool",
+            description="Tool with full control including _meta field",
+            inputSchema={
+                "type": "object",
+                "properties": {"message": {"type": "string"}},
+                "required": ["message"],
+            },
+        )
+    ]
+
+
+@server.call_tool()
+async def handle_call_tool(name: str, arguments: dict[str, Any]) -> types.CallToolResult:
+    """Handle tool calls by returning CallToolResult directly."""
+    if name == "advanced_tool":
+        message = str(arguments.get("message", ""))
+        return types.CallToolResult(
+            content=[types.TextContent(type="text", text=f"Processed: {message}")],
+            structuredContent={"result": "success", "message": message},
+            _meta={"hidden": "data for client applications only"},
+        )
+
+    raise ValueError(f"Unknown tool: {name}")
+
+
+async def run():
+    """Run the server."""
+    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
+        await server.run(
+            read_stream,
+            write_stream,
+            InitializationOptions(
+                server_name="example",
+                server_version="0.1.0",
+                capabilities=server.get_capabilities(
+                    notification_options=NotificationOptions(),
+                    experimental_capabilities={},
+                ),
+            ),
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(run())
diff --git a/src/mcp/server/fastmcp/utilities/func_metadata.py b/src/mcp/server/fastmcp/utilities/func_metadata.py
@@ -1,9 +1,10 @@
 import inspect
 import json
+import types
 from collections.abc import Awaitable, Callable, Sequence
 from itertools import chain
 from types import GenericAlias
-from typing import Annotated, Any, ForwardRef, cast, get_args, get_origin, get_type_hints
+from typing import Annotated, Any, ForwardRef, Union, cast, get_args, get_origin, get_type_hints
 
 import pydantic_core
 from pydantic import (
@@ -22,7 +23,7 @@
 from mcp.server.fastmcp.exceptions import InvalidSignature
 from mcp.server.fastmcp.utilities.logging import get_logger
 from mcp.server.fastmcp.utilities.types import Audio, Image
-from mcp.types import ContentBlock, TextContent
+from mcp.types import CallToolResult, ContentBlock, TextContent
 
 logger = get_logger(__name__)
 
@@ -104,6 +105,12 @@ def convert_result(self, result: Any) -> Any:
         from function return values, whereas the lowlevel server simply serializes
         the structured output.
         """
+        if isinstance(result, CallToolResult):
+            if self.output_schema is not None:
+                assert self.output_model is not None, "Output model must be set if output schema is defined"
+                self.output_model.model_validate(result.structuredContent)
+            return result
+
         unstructured_content = _convert_to_content(result)
 
         if self.output_schema is None:
@@ -268,6 +275,26 @@ def func_metadata(
     output_info = FieldInfo.from_annotation(_get_typed_annotation(sig.return_annotation, globalns))
     annotation = output_info.annotation
 
+    # Reject CallToolResult in Union types (including Optional)
+    # Handle both typing.Union (Union[X, Y]) and types.UnionType (X | Y)
+    origin = get_origin(annotation)
+    if origin is Union or origin is types.UnionType:
+        args = get_args(annotation)
+        # Check if CallToolResult appears in the union (excluding None for Optional check)
+        if any(isinstance(arg, type) and issubclass(arg, CallToolResult) for arg in args if arg is not type(None)):
+            raise InvalidSignature(
+                f"Function {func.__name__}: CallToolResult cannot be used in Union or Optional types. "
+                "To return empty results, use: CallToolResult(content=[])"
+            )
+
+    # if the typehint is CallToolResult, the user either intends to return without validation
+    # or they provided validation as Annotated metadata
+    if isinstance(annotation, type) and issubclass(annotation, CallToolResult):
+        if output_info.metadata:
+            annotation = output_info.metadata[0]
+        else:
+            return FuncMetadata(arg_model=arguments_model)
+
     output_model, output_schema, wrap_output = _try_create_model_and_schema(annotation, func.__name__, output_info)
 
     if output_model is None and structured_output is True:

diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py
@@ -480,7 +480,7 @@ def call_tool(self, *, validate_input: bool = True):
         def decorator(
             func: Callable[
                 ...,
-                Awaitable[UnstructuredContent | StructuredContent | CombinationContent],
+                Awaitable[UnstructuredContent | StructuredContent | CombinationContent | types.CallToolResult],
             ],
         ):
             logger.debug("Registering handler for CallToolRequest")
@@ -504,7 +504,9 @@ async def handler(req: types.CallToolRequest):
                     # output normalization
                     unstructured_content: UnstructuredContent
                     maybe_structured_content: StructuredContent | None
-                    if isinstance(results, tuple) and len(results) == 2:
+                    if isinstance(results, types.CallToolResult):
+                        return types.ServerResult(results)
+                    elif isinstance(results, tuple) and len(results) == 2:
                         # tool returned both structured and unstructured content
                         unstructured_content, maybe_structured_content = cast(CombinationContent, results)
                     elif isinstance(results, dict):