From db24c85359cef1a7c6bb78f192c475ddc5b289b3 Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Fri, 27 Jun 2025 21:41:37 -0400
Subject: [PATCH] Update all tests

---
 src/fastmcp/server/openapi.py               |   4 +-
 src/fastmcp/tools/tool.py                   |  66 +++--
 src/fastmcp/tools/tool_transform.py         |  41 ++-
 src/fastmcp/utilities/json_schema_type.py   |  50 +++-
 tests/contrib/test_bulk_tool_caller.py      |  18 +-
 tests/server/http/test_http_dependencies.py |  10 +-
 tests/server/test_import_server.py          |   2 +-
 tests/server/test_server.py                 |  75 +++---
 tests/server/test_server_interactions.py    | 194 +++++++++++++-
 tests/tools/test_tool.py                    | 256 +++++++++++++++++--
 tests/tools/test_tool_transform.py          | 270 +++++++++++++++++++-
 tests/utilities/test_json_schema_type.py    |  25 +-
 12 files changed, 911 insertions(+), 100 deletions(-)

diff --git a/src/fastmcp/server/openapi.py b/src/fastmcp/server/openapi.py
index 69a815261..1e8051ace 100644
--- a/src/fastmcp/server/openapi.py
+++ b/src/fastmcp/server/openapi.py
@@ -450,8 +450,10 @@ class OpenAPITool(Tool):
             # Try to parse as JSON first
             try:
                 result = response.json()
+                if not isinstance(result, dict):
+                    result = {"result": result}
                 return ToolResult(structured_content=result)
-            except (json.JSONDecodeError, ValueError):
+            except json.JSONDecodeError:
                 return ToolResult(content=response.text)
 
         except httpx.HTTPStatusError as e:
diff --git a/src/fastmcp/tools/tool.py b/src/fastmcp/tools/tool.py
index 9a53acb69..bd8d8a497 100644
--- a/src/fastmcp/tools/tool.py
+++ b/src/fastmcp/tools/tool.py
@@ -240,6 +240,13 @@ class FunctionTool(Tool):
             output_schema = None
         # Note: explicit schemas (dict) are used as-is without auto-wrapping
 
+        # Validate that explicit schemas are object type for structured content
+        if output_schema is not None and isinstance(output_schema, dict):
+            if output_schema.get("type") != "object":
+                raise ValueError(
+                    f'Output schemas must have "type" set to "object" due to MCP spec limitations. Received: {output_schema!r}'
+                )
+
         return cls(
             fn=parsed_fn.fn,
             name=name or parsed_fn.name,
@@ -264,6 +271,7 @@ class FunctionTool(Tool):
 
         type_adapter = get_cached_typeadapter(self.fn)
         result = type_adapter.validate_python(arguments)
+
         if inspect.isawaitable(result):
             result = await result
 
@@ -275,6 +283,7 @@ class FunctionTool(Tool):
         # Handle structured content based on output schema
         if self.output_schema is not None:
             if self.output_schema.get("x-fastmcp-wrap-result"):
+                # Schema says wrap - always wrap in result key
                 structured_output = {"result": result}
             else:
                 structured_output = result
@@ -354,26 +363,43 @@ class ParsedFunction:
         output_schema = None
         output_type = inspect.signature(fn).return_annotation
 
-        # there are a variety of types that we don't want to attempt to
-        # serialize because they are either used by FastMCP internally,
-        # or are MCP content types that explicitly don't form structured
-        # content. By replacing them with an explicitly unserializable type,
-        # we ensure that no output schema is automatically generated.
+        if output_type not in (inspect._empty, None, Any, ...):
+            # there are a variety of types that we don't want to attempt to
+            # serialize because they are either used by FastMCP internally,
+            # or are MCP content types that explicitly don't form structured
+            # content. By replacing them with an explicitly unserializable type,
+            # we ensure that no output schema is automatically generated.
+            output_type = replace_type(
+                output_type,
+                {
+                    t: _UnserializableType
+                    for t in (
+                        Image,
+                        Audio,
+                        File,
+                        ToolResult,
+                        mcp.types.TextContent,
+                        mcp.types.ImageContent,
+                        mcp.types.AudioContent,
+                        mcp.types.ResourceLink,
+                        mcp.types.EmbeddedResource,
+                    )
+                },
+            )
 
-        output_type = replace_type(
-            output_type,
-            {
-                inspect._empty: _UnserializableType,
-                Image: _UnserializableType,
-                Audio: _UnserializableType,
-                File: _UnserializableType,
-                ToolResult: _UnserializableType,
-                mcp.types.TextContent: _UnserializableType,
-                mcp.types.ImageContent: _UnserializableType,
-                mcp.types.AudioContent: _UnserializableType,
-                mcp.types.ResourceLink: _UnserializableType,
-                mcp.types.EmbeddedResource: _UnserializableType,
-            },
+            try:
+                output_type_adapter = get_cached_typeadapter(output_type)
+                output_schema = output_type_adapter.json_schema()
+            except PydanticSchemaGenerationError as e:
+                if "_UnserializableType" not in str(e):
+                    logger.debug(f"Unable to generate schema for type {output_type!r}")
+
+        return cls(
+            fn=fn,
+            name=fn_name,
+            description=fn_doc,
+            input_schema=input_schema,
+            output_schema=output_schema or None,
         )
 
         try:
@@ -388,7 +414,7 @@ class ParsedFunction:
             name=fn_name,
             description=fn_doc,
             input_schema=input_schema,
-            output_schema=output_schema,
+            output_schema=output_schema or None,
         )
 
 
diff --git a/src/fastmcp/tools/tool_transform.py b/src/fastmcp/tools/tool_transform.py
index f52a54da3..db842c519 100644
--- a/src/fastmcp/tools/tool_transform.py
+++ b/src/fastmcp/tools/tool_transform.py
@@ -269,9 +269,32 @@ class TransformedTool(Tool):
         try:
             result = await self.fn(**arguments)
 
-            # If transform function returns ToolResult, use it directly
+            # If transform function returns ToolResult, respect our output_schema setting
             if isinstance(result, ToolResult):
-                return result
+                if self.output_schema is None:
+                    # Check if this is from a custom function that returns ToolResult
+                    import inspect
+
+                    return_annotation = inspect.signature(self.fn).return_annotation
+                    if return_annotation is ToolResult:
+                        # Custom function returns ToolResult - preserve its content
+                        return result
+                    else:
+                        # Forwarded call with disabled schema - strip structured content
+                        return ToolResult(
+                            content=result.content,
+                            structured_content=None,
+                        )
+                elif self.output_schema.get(
+                    "type"
+                ) != "object" and not self.output_schema.get("x-fastmcp-wrap-result"):
+                    # Non-object explicit schemas disable structured content
+                    return ToolResult(
+                        content=result.content,
+                        structured_content=None,
+                    )
+                else:
+                    return result
 
             # Otherwise convert to content and create ToolResult with proper structured content
             from fastmcp.tools.tool import _convert_to_content
@@ -283,8 +306,11 @@ class TransformedTool(Tool):
             # Handle structured content based on output schema
             if self.output_schema is not None:
                 if self.output_schema.get("x-fastmcp-wrap-result"):
+                    # Schema says wrap - always wrap in result key
                     structured_output = {"result": result}
                 else:
+                    # Object schemas - use result directly
+                    # User is responsible for returning dict-compatible data
                     structured_output = result
             else:
                 structured_output = None
@@ -381,7 +407,16 @@ class TransformedTool(Tool):
                 parsed_fn = ParsedFunction.from_function(transform_fn, validate=False)
                 final_output_schema = _wrap_schema_if_needed(parsed_fn.output_schema)
                 if final_output_schema is None:
-                    final_output_schema = tool.output_schema
+                    # Check if function returns ToolResult - if so, don't fall back to parent
+                    import inspect
+
+                    return_annotation = inspect.signature(
+                        transform_fn
+                    ).return_annotation
+                    if return_annotation is ToolResult:
+                        final_output_schema = None
+                    else:
+                        final_output_schema = tool.output_schema
             else:
                 final_output_schema = tool.output_schema
 
diff --git a/src/fastmcp/utilities/json_schema_type.py b/src/fastmcp/utilities/json_schema_type.py
index 2e48cd8a0..6be3c49ad 100644
--- a/src/fastmcp/utilities/json_schema_type.py
+++ b/src/fastmcp/utilities/json_schema_type.py
@@ -169,8 +169,17 @@ def json_schema_to_type(
     """
     # Always use the top-level schema for references
     if schema.get("type") == "object":
-        # If no properties defined but additionalProperties is True, return dict[str, Any]
-        if not schema.get("properties") and schema.get("additionalProperties") is True:
+        # If no properties defined but has additionalProperties, return typed dict
+        if not schema.get("properties") and schema.get("additionalProperties"):
+            additional_props = schema["additionalProperties"]
+            if additional_props is True:
+                return dict[str, Any]  # type: ignore - additionalProperties: true means dict[str, Any]
+            else:
+                # Handle typed dictionaries like dict[str, str]
+                value_type = _schema_to_type(additional_props, schemas=schema)
+                return dict[str, value_type]  # type: ignore
+        # If no properties and no additionalProperties, default to dict[str, Any] for safety
+        elif not schema.get("properties") and not schema.get("additionalProperties"):
             return dict[str, Any]  # type: ignore
         # If has properties AND additionalProperties is True, use Pydantic BaseModel
         elif schema.get("properties") and schema.get("additionalProperties") is True:
@@ -328,6 +337,43 @@ def _schema_to_type(
     if "enum" in schema:
         return _create_enum(f"Enum_{len(_classes)}", schema["enum"])
 
+    # Handle anyOf unions
+    if "anyOf" in schema:
+        types: list[type | Any] = []
+        for subschema in schema["anyOf"]:
+            # Special handling for dict-like objects in unions
+            if (
+                subschema.get("type") == "object"
+                and not subschema.get("properties")
+                and subschema.get("additionalProperties")
+            ):
+                # This is a dict type, handle it directly
+                additional_props = subschema["additionalProperties"]
+                if additional_props is True:
+                    types.append(dict[str, Any])  # type: ignore
+                else:
+                    value_type = _schema_to_type(additional_props, schemas)
+                    types.append(dict[str, value_type])  # type: ignore
+            else:
+                types.append(_schema_to_type(subschema, schemas))
+
+        # Check if one of the types is None (null)
+        has_null = type(None) in types
+        types = [t for t in types if t is not type(None)]
+
+        if len(types) == 0:
+            return type(None)
+        elif len(types) == 1:
+            if has_null:
+                return Optional[types[0]]  # type: ignore # noqa: UP007
+            else:
+                return types[0]
+        else:
+            if has_null:
+                return Union[tuple(types + [type(None)])]  # type: ignore # noqa: UP007
+            else:
+                return Union[tuple(types)]  # type: ignore # noqa: UP007
+
     schema_type = schema.get("type")
     if not schema_type:
         return Any
diff --git a/tests/contrib/test_bulk_tool_caller.py b/tests/contrib/test_bulk_tool_caller.py
index 38dc16872..578dcba96 100644
--- a/tests/contrib/test_bulk_tool_caller.py
+++ b/tests/contrib/test_bulk_tool_caller.py
@@ -45,13 +45,8 @@ async def echo_tool(arg1: str) -> str:
 def echo_tool_result_factory(arg1: str) -> CallToolRequestResult:
     """A tool that returns a result based on the input arguments."""
     return CallToolRequestResult(
-        isError=True,
-        content=[
-            TextContent(
-                text="Output validation error: outputSchema defined but no structured output returned",
-                type="text",
-            )
-        ],
+        isError=False,
+        content=[TextContent(text=f"{arg1}", type="text")],
         tool="echo_tool",
         arguments={"arg1": arg1},
     )
@@ -64,13 +59,8 @@ async def no_return_tool(arg1: str) -> None:
 def no_return_tool_result_factory(arg1: str) -> CallToolRequestResult:
     """A tool that returns a result based on the input arguments."""
     return CallToolRequestResult(
-        isError=True,
-        content=[
-            TextContent(
-                text="Output validation error: outputSchema defined but no structured output returned",
-                type="text",
-            )
-        ],
+        isError=False,
+        content=[],
         tool="no_return_tool",
         arguments={"arg1": arg1},
     )
diff --git a/tests/server/http/test_http_dependencies.py b/tests/server/http/test_http_dependencies.py
index ff88e1e7a..32aa87588 100644
--- a/tests/server/http/test_http_dependencies.py
+++ b/tests/server/http/test_http_dependencies.py
@@ -86,9 +86,8 @@ async def test_http_headers_tool_shttp(shttp_server: str):
         )
     ) as client:
         result = await client.call_tool("get_headers_tool")
-        json_result = json.loads(result[0].text)  # type: ignore[attr-defined]
-        assert "x-demo-header" in json_result
-        assert json_result["x-demo-header"] == "ABC"
+        assert "x-demo-header" in result.data
+        assert result.data["x-demo-header"] == "ABC"
 
 
 async def test_http_headers_tool_sse(sse_server: str):
@@ -96,9 +95,8 @@ async def test_http_headers_tool_sse(sse_server: str):
         transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
     ) as client:
         result = await client.call_tool("get_headers_tool")
-        json_result = json.loads(result[0].text)  # type: ignore[attr-defined]
-        assert "x-demo-header" in json_result
-        assert json_result["x-demo-header"] == "ABC"
+        assert "x-demo-header" in result.data
+        assert result.data["x-demo-header"] == "ABC"
 
 
 async def test_http_headers_prompt_shttp(shttp_server: str):
diff --git a/tests/server/test_import_server.py b/tests/server/test_import_server.py
index fcabb1ac9..1f8f6611c 100644
--- a/tests/server/test_import_server.py
+++ b/tests/server/test_import_server.py
@@ -278,7 +278,7 @@ async def test_call_nested_imported_tool():
 
     async with Client(main_app) as client:
         result = await client.call_tool("service_provider_compute", {"input": 21})
-        assert result.data == "42"
+        assert result.data == 42
 
 
 async def test_import_with_proxy_tools():
diff --git a/tests/server/test_server.py b/tests/server/test_server.py
index 59e4bdc07..9c08955c2 100644
--- a/tests/server/test_server.py
+++ b/tests/server/test_server.py
@@ -127,8 +127,9 @@ class TestToolDecorator:
         def add(x: int, y: int) -> int:
             return x + y
 
-        result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
-        assert result[0].text == "3"  # type: ignore[attr-defined]
+        async with Client(mcp) as client:
+            result = await client.call_tool("add", {"x": 1, "y": 2})
+            assert result.data == 3
 
     async def test_tool_decorator_without_parentheses(self):
         """Test that @tool decorator works without parentheses."""
@@ -144,8 +145,9 @@ class TestToolDecorator:
         assert "add" in tools
 
         # Verify it can be called
-        result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
-        assert result[0].text == "3"  # type: ignore[attr-defined]
+        async with Client(mcp) as client:
+            result = await client.call_tool("add", {"x": 1, "y": 2})
+            assert result.data == 3
 
     async def test_tool_decorator_with_name(self):
         mcp = FastMCP()
@@ -154,8 +156,9 @@ class TestToolDecorator:
         def add(x: int, y: int) -> int:
             return x + y
 
-        result = await mcp._mcp_call_tool("custom-add", {"x": 1, "y": 2})
-        assert result[0].text == "3"  # type: ignore[attr-defined]
+        async with Client(mcp) as client:
+            result = await client.call_tool("custom-add", {"x": 1, "y": 2})
+            assert result.data == 3
 
     async def test_tool_decorator_with_description(self):
         mcp = FastMCP()
@@ -181,8 +184,9 @@ class TestToolDecorator:
 
         obj = MyClass(10)
         mcp.add_tool(Tool.from_function(obj.add))
-        result = await mcp._mcp_call_tool("add", {"y": 2})
-        assert result[0].text == "12"  # type: ignore[attr-defined]
+        async with Client(mcp) as client:
+            result = await client.call_tool("add", {"y": 2})
+            assert result.data == 12
 
     async def test_tool_decorator_classmethod(self):
         mcp = FastMCP()
@@ -195,8 +199,9 @@ class TestToolDecorator:
                 return cls.x + y
 
         mcp.add_tool(Tool.from_function(MyClass.add))
-        result = await mcp._mcp_call_tool("add", {"y": 2})
-        assert result[0].text == "12"  # type: ignore[attr-defined]
+        async with Client(mcp) as client:
+            result = await client.call_tool("add", {"y": 2})
+            assert result.data == 12
 
     async def test_tool_decorator_staticmethod(self):
         mcp = FastMCP()
@@ -207,8 +212,9 @@ class TestToolDecorator:
             def add(x: int, y: int) -> int:
                 return x + y
 
-        result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
-        assert result[0].text == "3"  # type: ignore[attr-defined]
+        async with Client(mcp) as client:
+            result = await client.call_tool("add", {"x": 1, "y": 2})
+            assert result.data == 3
 
     async def test_tool_decorator_async_function(self):
         mcp = FastMCP()
@@ -217,8 +223,9 @@ class TestToolDecorator:
         async def add(x: int, y: int) -> int:
             return x + y
 
-        result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
-        assert result[0].text == "3"  # type: ignore[attr-defined]
+        async with Client(mcp) as client:
+            result = await client.call_tool("add", {"x": 1, "y": 2})
+            assert result.data == 3
 
     async def test_tool_decorator_classmethod_error(self):
         mcp = FastMCP()
@@ -242,8 +249,9 @@ class TestToolDecorator:
                 return cls.x + y
 
         mcp.add_tool(Tool.from_function(MyClass.add))
-        result = await mcp._mcp_call_tool("add", {"y": 2})
-        assert result[0].text == "12"  # type: ignore[attr-defined]
+        async with Client(mcp) as client:
+            result = await client.call_tool("add", {"y": 2})
+            assert result.data == 12
 
     async def test_tool_decorator_staticmethod_async_function(self):
         mcp = FastMCP()
@@ -254,8 +262,9 @@ class TestToolDecorator:
                 return x + y
 
         mcp.add_tool(Tool.from_function(MyClass.add))
-        result = await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
-        assert result[0].text == "3"  # type: ignore[attr-defined]
+        async with Client(mcp) as client:
+            result = await client.call_tool("add", {"x": 1, "y": 2})
+            assert result.data == 3
 
     async def test_tool_decorator_staticmethod_order(self):
         """Test that the recommended decorator order works for static methods"""
@@ -268,8 +277,9 @@ class TestToolDecorator:
                 return x + y
 
         # Test that the recommended order works
-        result = await mcp._mcp_call_tool("add_v1", {"x": 1, "y": 2})
-        assert result[0].text == "3"  # type: ignore[attr-defined]
+        async with Client(mcp) as client:
+            result = await client.call_tool("add_v1", {"x": 1, "y": 2})
+            assert result.data == 3
 
     async def test_tool_decorator_with_tags(self):
         """Test that the tool decorator properly sets tags."""
@@ -299,8 +309,9 @@ class TestToolDecorator:
         assert "custom_multiply" in tools
 
         # Call the tool by its custom name
-        result = await mcp._mcp_call_tool("custom_multiply", {"a": 5, "b": 3})
-        assert result[0].text == "15"  # type: ignore[attr-defined]
+        async with Client(mcp) as client:
+            result = await client.call_tool("custom_multiply", {"a": 5, "b": 3})
+            assert result.data == 15
 
         # Original name should not be registered
         assert "multiply" not in tools
@@ -354,8 +365,9 @@ class TestToolDecorator:
         assert tools["direct_call_tool"] is result_fn
 
         # Verify it can be called
-        result = await mcp._mcp_call_tool("direct_call_tool", {"x": 5, "y": 3})
-        assert result[0].text == "8"  # type: ignore[attr-defined]
+        async with Client(mcp) as client:
+            result = await client.call_tool("direct_call_tool", {"x": 5, "y": 3})
+            assert result.data == 8
 
     async def test_tool_decorator_with_string_name(self):
         """Test that @tool("custom_name") syntax works correctly."""
@@ -372,8 +384,9 @@ class TestToolDecorator:
         assert "my_function" not in tools  # Original name should not be registered
 
         # Verify it can be called
-        result = await mcp._mcp_call_tool("string_named_tool", {"x": 42})
-        assert result[0].text == "Result: 42"  # type: ignore[attr-defined]
+        async with Client(mcp) as client:
+            result = await client.call_tool("string_named_tool", {"x": 42})
+            assert result.data == "Result: 42"
 
     async def test_tool_decorator_conflicting_names_error(self):
         """Test that providing both positional and keyword name raises an error."""
@@ -391,11 +404,13 @@ class TestToolDecorator:
     async def test_tool_decorator_with_output_schema(self):
         mcp = FastMCP()
 
-        @mcp.tool(output_schema={"type": "integer"})
-        def my_function(x: int) -> str:
-            return f"Result: {x}"
+        with pytest.raises(
+            ValueError, match='Output schemas must have "type" set to "object"'
+        ):
 
-        assert my_function.output_schema == {"type": "integer"}
+            @mcp.tool(output_schema={"type": "integer"})
+            def my_function(x: int) -> str:
+                return f"Result: {x}"
 
 
 class TestResourceDecorator:
diff --git a/tests/server/test_server_interactions.py b/tests/server/test_server_interactions.py
index 1e810d891..b91b83b44 100644
--- a/tests/server/test_server_interactions.py
+++ b/tests/server/test_server_interactions.py
@@ -5,7 +5,7 @@ import uuid
 from dataclasses import dataclass
 from enum import Enum
 from pathlib import Path
-from typing import Annotated, Literal
+from typing import Annotated, Any, Literal
 
 import pytest
 from mcp import McpError
@@ -946,6 +946,198 @@ class TestToolOutputSchema:
             assert result.structured_content == {"message": "Hello, world!"}
             assert result.data == {"message": "Hello, world!"}
 
+    async def test_output_schema_false_full_handshake(self):
+        """Test that output_schema=False works through full client/server handshake."""
+        mcp = FastMCP()
+
+        @mcp.tool(output_schema=False)
+        def simple_tool() -> dict[str, str]:
+            return {"message": "Hello from disabled schema"}
+
+        async with Client(mcp) as client:
+            # List tools and verify output schema is None
+            tools = await client.list_tools()
+            tool = next(t for t in tools if t.name == "simple_tool")
+            assert tool.outputSchema is None
+
+            # Call tool and verify no structured content
+            result = await client.call_tool("simple_tool", {})
+            assert result.structured_content is None
+            assert result.data is None
+            assert json.loads(result.content[0].text) == {
+                "message": "Hello from disabled schema"
+            }  # type: ignore[attr-defined]
+
+    async def test_output_schema_explicit_object_full_handshake(self):
+        """Test explicit object output schema through full client/server handshake."""
+        mcp = FastMCP()
+
+        @mcp.tool(
+            output_schema={
+                "type": "object",
+                "properties": {
+                    "greeting": {"type": "string"},
+                    "count": {"type": "integer"},
+                },
+                "required": ["greeting"],
+            }
+        )
+        def explicit_tool() -> dict[str, Any]:
+            return {"greeting": "Hello", "count": 42}
+
+        async with Client(mcp) as client:
+            # List tools and verify exact schema is preserved
+            tools = await client.list_tools()
+            tool = next(t for t in tools if t.name == "explicit_tool")
+            expected_schema = {
+                "type": "object",
+                "properties": {
+                    "greeting": {"type": "string"},
+                    "count": {"type": "integer"},
+                },
+                "required": ["greeting"],
+            }
+            assert tool.outputSchema == expected_schema
+
+            # Call tool and verify structured content matches return value directly
+            result = await client.call_tool("explicit_tool", {})
+            assert result.structured_content == {"greeting": "Hello", "count": 42}
+            # Client deserializes according to schema, so check fields
+            assert result.data.greeting == "Hello"  # type: ignore[attr-defined]
+            assert result.data.count == 42  # type: ignore[attr-defined]
+
+    async def test_output_schema_wrapped_primitive_full_handshake(self):
+        """Test wrapped primitive output schema through full client/server handshake."""
+        mcp = FastMCP()
+
+        @mcp.tool
+        def primitive_tool() -> str:
+            return "Hello, primitives!"
+
+        async with Client(mcp) as client:
+            # List tools and verify schema shows wrapped structure
+            tools = await client.list_tools()
+            tool = next(t for t in tools if t.name == "primitive_tool")
+            expected_schema = {
+                "type": "object",
+                "properties": {"result": {"type": "string"}},
+                "x-fastmcp-wrap-result": True,
+            }
+            assert tool.outputSchema == expected_schema
+
+            # Call tool and verify structured content is wrapped
+            result = await client.call_tool("primitive_tool", {})
+            assert result.structured_content == {"result": "Hello, primitives!"}
+            assert result.data == "Hello, primitives!"  # Client unwraps for convenience
+
+    async def test_output_schema_complex_type_full_handshake(self):
+        """Test complex type output schema through full client/server handshake."""
+        mcp = FastMCP()
+
+        @mcp.tool
+        def complex_tool() -> list[dict[str, int]]:
+            return [{"a": 1, "b": 2}, {"c": 3, "d": 4}]
+
+        async with Client(mcp) as client:
+            # List tools and verify schema shows wrapped array
+            tools = await client.list_tools()
+            tool = next(t for t in tools if t.name == "complex_tool")
+            expected_inner_schema = TypeAdapter(list[dict[str, int]]).json_schema()
+            expected_schema = {
+                "type": "object",
+                "properties": {"result": expected_inner_schema},
+                "x-fastmcp-wrap-result": True,
+            }
+            assert tool.outputSchema == expected_schema
+
+            # Call tool and verify structured content is wrapped
+            result = await client.call_tool("complex_tool", {})
+            expected_data = [{"a": 1, "b": 2}, {"c": 3, "d": 4}]
+            assert result.structured_content == {"result": expected_data}
+            # Client deserializes - just verify we got data back
+            assert result.data is not None
+
+    async def test_output_schema_dataclass_full_handshake(self):
+        """Test dataclass output schema through full client/server handshake."""
+        mcp = FastMCP()
+
+        @dataclass
+        class User:
+            name: str
+            age: int
+
+        @mcp.tool
+        def dataclass_tool() -> User:
+            return User(name="Alice", age=30)
+
+        async with Client(mcp) as client:
+            # List tools and verify schema is object type (not wrapped)
+            tools = await client.list_tools()
+            tool = next(t for t in tools if t.name == "dataclass_tool")
+            expected_schema = TypeAdapter(User).json_schema()
+            assert tool.outputSchema == expected_schema
+            assert "x-fastmcp-wrap-result" not in tool.outputSchema
+
+            # Call tool and verify structured content is direct
+            result = await client.call_tool("dataclass_tool", {})
+            assert result.structured_content == {"name": "Alice", "age": 30}
+            # Client deserializes according to schema
+            assert result.data.name == "Alice"  # type: ignore[attr-defined]
+            assert result.data.age == 30  # type: ignore[attr-defined]
+
+    async def test_output_schema_mixed_content_types(self):
+        """Test tools with mixed content and output schemas."""
+        mcp = FastMCP()
+
+        @mcp.tool
+        def mixed_output() -> list[Any]:
+            # Return mixed content that includes MCP types and regular data
+            return [
+                "text message",
+                {"structured": "data"},
+                TextContent(type="text", text="direct MCP content"),
+            ]
+
+        async with Client(mcp) as client:
+            result = await client.call_tool("mixed_output", {})
+
+            # Should have multiple content blocks
+            assert len(result.content) >= 2
+
+            # Should have structured output with wrapped result
+            expected_data = [
+                "text message",
+                {"structured": "data"},
+                {
+                    "type": "text",
+                    "text": "direct MCP content",
+                    "annotations": None,
+                    "_meta": None,
+                },
+            ]
+            assert result.structured_content == {"result": expected_data}
+
+    async def test_output_schema_serialization_edge_cases(self):
+        """Test edge cases in output schema serialization."""
+        mcp = FastMCP()
+
+        @mcp.tool
+        def edge_case_tool() -> tuple[int, str]:
+            return (42, "hello")
+
+        async with Client(mcp) as client:
+            # Verify tuple gets proper schema
+            tools = await client.list_tools()
+            tool = next(t for t in tools if t.name == "edge_case_tool")
+
+            # Tuples should be wrapped since they're not object type
+            assert "x-fastmcp-wrap-result" in tool.outputSchema
+
+            result = await client.call_tool("edge_case_tool", {})
+            # Should be wrapped with result key
+            assert result.structured_content == {"result": [42, "hello"]}
+            assert result.data == [42, "hello"]
+
 
 class TestToolContextInjection:
     """Test context injection in tools."""
diff --git a/tests/tools/test_tool.py b/tests/tools/test_tool.py
index b03559cdf..6a4b5ac24 100644
--- a/tests/tools/test_tool.py
+++ b/tests/tools/test_tool.py
@@ -263,14 +263,16 @@ class TestToolFromFunctionOutputSchema:
     @pytest.mark.parametrize(
         "annotation",
         [
-            None,
             int,
             float,
             bool,
             str,
             int | float,
+            list,
             list[int],
             list[int | float],
+            dict,
+            dict[str, Any],
             dict[str, int | None],
             tuple[int, str],
             set[int],
@@ -304,7 +306,6 @@ class TestToolFromFunctionOutputSchema:
     @pytest.mark.parametrize(
         "annotation",
         [
-            Any,
             AnyUrl,
             Annotated[int, Field(ge=1)],
             Annotated[int, Field(ge=1)],
@@ -317,17 +318,26 @@ class TestToolFromFunctionOutputSchema:
         tool = Tool.from_function(func)
         base_schema = TypeAdapter(annotation).json_schema()
 
-        # Special case for Any type - it generates an empty schema and doesn't get wrapped
-        if annotation is Any:
-            assert tool.output_schema == base_schema  # Should be {}
-        else:
-            # All other non-object types get wrapped, including complex constrained types
-            expected_schema = {
-                "type": "object",
-                "properties": {"result": base_schema},
-                "x-fastmcp-wrap-result": True,
-            }
-            assert tool.output_schema == expected_schema
+        expected_schema = {
+            "type": "object",
+            "properties": {"result": base_schema},
+            "x-fastmcp-wrap-result": True,
+        }
+        assert tool.output_schema == expected_schema
+
+    async def test_none_return_annotation(self):
+        def func() -> None:
+            pass
+
+        tool = Tool.from_function(func)
+        assert tool.output_schema is None
+
+    async def test_any_return_annotation(self):
+        def func() -> Any:
+            return 1
+
+        tool = Tool.from_function(func)
+        assert tool.output_schema is None
 
     @pytest.mark.parametrize(
         "annotation, expected",
@@ -413,7 +423,7 @@ class TestToolFromFunctionOutputSchema:
             return {"a": 1, "b": 2}
 
         # Provide a custom output schema that differs from the inferred one
-        custom_schema = {"type": "string", "description": "Custom schema"}
+        custom_schema = {"type": "object", "description": "Custom schema"}
 
         tool = Tool.from_function(func, output_schema=custom_schema)
         assert tool.output_schema == custom_schema
@@ -445,7 +455,10 @@ class TestToolFromFunctionOutputSchema:
             return Unserializable(data="test")
 
         # Provide a custom output schema even though the annotation is unserializable
-        custom_schema = {"type": "array", "items": {"type": "string"}}
+        custom_schema = {
+            "type": "object",
+            "properties": {"items": {"type": "array", "items": {"type": "string"}}},
+        }
 
         tool = Tool.from_function(func, output_schema=custom_schema)
         assert tool.output_schema == custom_schema
@@ -457,7 +470,10 @@ class TestToolFromFunctionOutputSchema:
             return "hello"
 
         # Provide a custom output schema even though there's no return annotation
-        custom_schema = {"type": "number", "minimum": 0}
+        custom_schema = {
+            "type": "object",
+            "properties": {"value": {"type": "number", "minimum": 0}},
+        }
 
         tool = Tool.from_function(func, output_schema=custom_schema)
         assert tool.output_schema == custom_schema
@@ -486,7 +502,7 @@ class TestToolFromFunctionOutputSchema:
             return "hello"
 
         # Provide a custom output schema that differs from the inferred union schema
-        custom_schema = {"type": "boolean"}
+        custom_schema = {"type": "object", "properties": {"flag": {"type": "boolean"}}}
 
         tool = Tool.from_function(func, output_schema=custom_schema)
         assert tool.output_schema == custom_schema
@@ -504,11 +520,215 @@ class TestToolFromFunctionOutputSchema:
             return Person(name="John", age=30)
 
         # Provide a custom output schema that differs from the inferred Person schema
-        custom_schema = {"type": "array", "items": {"type": "number"}}
+        custom_schema = {
+            "type": "object",
+            "properties": {"numbers": {"type": "array", "items": {"type": "number"}}},
+        }
 
         tool = Tool.from_function(func, output_schema=custom_schema)
         assert tool.output_schema == custom_schema
 
+    async def test_output_schema_false_disables_structured_content(self):
+        """Test that output_schema=False disables structured content generation."""
+
+        def func() -> dict[str, str]:
+            return {"message": "Hello, world!"}
+
+        tool = Tool.from_function(func, output_schema=False)
+        assert tool.output_schema is None
+
+        result = await tool.run({})
+        assert result.structured_content is None
+        assert len(result.content) == 1
+        assert result.content[0].text == '{\n  "message": "Hello, world!"\n}'
+
+    async def test_output_schema_none_disables_structured_content(self):
+        """Test that output_schema=None explicitly disables structured content."""
+
+        def func() -> int:
+            return 42
+
+        tool = Tool.from_function(func, output_schema=None)
+        assert tool.output_schema is None
+
+        result = await tool.run({})
+        assert result.structured_content is None
+        assert len(result.content) == 1
+        assert result.content[0].text == "42"
+
+    async def test_output_schema_inferred_when_not_specified(self):
+        """Test that output schema is inferred when not explicitly specified."""
+
+        def func() -> int:
+            return 42
+
+        # Don't specify output_schema - should infer and wrap
+        tool = Tool.from_function(func)
+        expected_schema = {
+            "type": "object",
+            "properties": {"result": {"type": "integer"}},
+            "x-fastmcp-wrap-result": True,
+        }
+        assert tool.output_schema == expected_schema
+
+        result = await tool.run({})
+        assert result.structured_content == {"result": 42}
+
+    async def test_explicit_object_schema_with_dict_return(self):
+        """Test that explicit object schemas work when function returns a dict."""
+
+        def func() -> dict[str, int]:
+            return {"value": 42}
+
+        # Provide explicit object schema
+        explicit_schema = {
+            "type": "object",
+            "properties": {"value": {"type": "integer", "minimum": 0}},
+        }
+        tool = Tool.from_function(func, output_schema=explicit_schema)
+        assert tool.output_schema == explicit_schema  # Schema not wrapped
+        assert "x-fastmcp-wrap-result" not in tool.output_schema
+
+        result = await tool.run({})
+        # Dict result with object schema is used directly
+        assert result.structured_content == {"value": 42}
+        assert result.content[0].text == '{\n  "value": 42\n}'
+
+    async def test_explicit_object_schema_with_non_dict_return_fails(self):
+        """Test that explicit object schemas fail when function returns non-dict."""
+
+        def func() -> int:
+            return 42
+
+        # Provide explicit object schema but return non-dict
+        explicit_schema = {
+            "type": "object",
+            "properties": {"value": {"type": "integer"}},
+        }
+        tool = Tool.from_function(func, output_schema=explicit_schema)
+
+        # Should fail because int is not dict-compatible with object schema
+        with pytest.raises(ValueError, match="structured_content must be a dict"):
+            await tool.run({})
+
+    async def test_object_output_schema_not_wrapped(self):
+        """Test that object-type output schemas are never wrapped."""
+
+        def func() -> dict[str, int]:
+            return {"value": 42}
+
+        # Object schemas should never be wrapped, even when inferred
+        tool = Tool.from_function(func)
+        expected_schema = TypeAdapter(dict[str, int]).json_schema()
+        assert tool.output_schema == expected_schema  # Not wrapped
+        assert "x-fastmcp-wrap-result" not in tool.output_schema
+
+        result = await tool.run({})
+        assert result.structured_content == {"value": 42}  # Direct value
+
+    async def test_structured_content_interaction_with_wrapping(self):
+        """Test that structured content works correctly with schema wrapping."""
+
+        def func() -> str:
+            return "hello"
+
+        # Inferred schema should wrap string type
+        tool = Tool.from_function(func)
+        expected_schema = {
+            "type": "object",
+            "properties": {"result": {"type": "string"}},
+            "x-fastmcp-wrap-result": True,
+        }
+        assert tool.output_schema == expected_schema
+
+        result = await tool.run({})
+        # Unstructured content
+        assert len(result.content) == 1
+        assert result.content[0].text == "hello"
+        # Structured content should be wrapped
+        assert result.structured_content == {"result": "hello"}
+
+    async def test_structured_content_with_explicit_object_schema(self):
+        """Test structured content with explicit object schema."""
+
+        def func() -> dict[str, str]:
+            return {"greeting": "hello"}
+
+        # Provide explicit object schema
+        explicit_schema = {
+            "type": "object",
+            "properties": {"greeting": {"type": "string"}},
+            "required": ["greeting"],
+        }
+        tool = Tool.from_function(func, output_schema=explicit_schema)
+        assert tool.output_schema == explicit_schema
+
+        result = await tool.run({})
+        # Should use direct value since explicit schema doesn't have wrap marker
+        assert result.structured_content == {"greeting": "hello"}
+
+    async def test_structured_content_with_custom_wrapper_schema(self):
+        """Test structured content with custom schema that includes wrap marker."""
+
+        def func() -> str:
+            return "world"
+
+        # Custom schema with wrap marker
+        custom_schema = {
+            "type": "object",
+            "properties": {"message": {"type": "string"}},
+            "x-fastmcp-wrap-result": True,
+        }
+        tool = Tool.from_function(func, output_schema=custom_schema)
+        assert tool.output_schema == custom_schema
+
+        result = await tool.run({})
+        # Should wrap with "result" key due to wrap marker
+        assert result.structured_content == {"result": "world"}
+
+    async def test_none_vs_false_output_schema_behavior(self):
+        """Test the difference between None and False for output_schema."""
+
+        def func() -> int:
+            return 123
+
+        # None should disable
+        tool_none = Tool.from_function(func, output_schema=None)
+        assert tool_none.output_schema is None
+
+        # False should also disable
+        tool_false = Tool.from_function(func, output_schema=False)
+        assert tool_false.output_schema is None
+
+        # Both should have same behavior
+        result_none = await tool_none.run({})
+        result_false = await tool_false.run({})
+
+        assert result_none.structured_content is None
+        assert result_false.structured_content is None
+        assert result_none.content[0].text == result_false.content[0].text == "123"
+
+    async def test_non_object_output_schema_raises_error(self):
+        """Test that providing a non-object output schema raises a ValueError."""
+
+        def func() -> int:
+            return 42
+
+        # Test various non-object schemas that should raise errors
+        non_object_schemas = [
+            {"type": "string"},
+            {"type": "integer", "minimum": 0},
+            {"type": "number"},
+            {"type": "boolean"},
+            {"type": "array", "items": {"type": "string"}},
+        ]
+
+        for schema in non_object_schemas:
+            with pytest.raises(
+                ValueError, match='Output schemas must have "type" set to "object"'
+            ):
+                Tool.from_function(func, output_schema=schema)
+
 
 class TestConvertResultToContent:
     """Tests for the _convert_to_content helper function."""
diff --git a/tests/tools/test_tool_transform.py b/tests/tools/test_tool_transform.py
index 014422452..ba25d19a4 100644
--- a/tests/tools/test_tool_transform.py
+++ b/tests/tools/test_tool_transform.py
@@ -4,14 +4,15 @@ from typing import Annotated, Any
 
 import pytest
 from dirty_equals import IsList
-from pydantic import BaseModel, Field
+from mcp.types import TextContent
+from pydantic import BaseModel, Field, TypeAdapter
 from typing_extensions import TypedDict
 
 from fastmcp import FastMCP
 from fastmcp.client.client import Client
 from fastmcp.exceptions import ToolError
 from fastmcp.tools import Tool, forward, forward_raw
-from fastmcp.tools.tool import FunctionTool
+from fastmcp.tools.tool import FunctionTool, ToolResult
 from fastmcp.tools.tool_transform import ArgTransform, TransformedTool
 
 
@@ -691,7 +692,7 @@ async def test_arg_transform_type_precedence_runtime():
         # Convert string back to int for the original function
         result = await forward_raw(x=int(x), y=y)
         # Extract the text from the result
-        result_text = result.content[0].text
+        result_text = result.content[0].text  # type: ignore[attr-defined]
         return f"String input '{x}' converted to result: {result_text}"
 
     tool = Tool.from_tool(
@@ -1030,3 +1031,266 @@ def test_arg_transform_examples_in_schema(add_tool):
     )
     prop3 = get_property(new_tool3, "old_x")
     assert "examples" not in prop3
+
+
+class TestTransformToolOutputSchema:
+    """Test output schema handling in transformed tools."""
+
+    @pytest.fixture
+    def base_string_tool(self) -> FunctionTool:
+        """Tool that returns a string (gets wrapped)."""
+
+        def string_tool(x: int) -> str:
+            return f"Result: {x}"
+
+        return Tool.from_function(string_tool)
+
+    @pytest.fixture
+    def base_dict_tool(self) -> FunctionTool:
+        """Tool that returns a dict (object type, not wrapped)."""
+
+        def dict_tool(x: int) -> dict[str, int]:
+            return {"value": x}
+
+        return Tool.from_function(dict_tool)
+
+    def test_transform_inherits_parent_output_schema(self, base_string_tool):
+        """Test that transformed tool inherits parent's output schema by default."""
+        new_tool = Tool.from_tool(base_string_tool)
+
+        # Should inherit parent's wrapped string schema
+        expected_schema = {
+            "type": "object",
+            "properties": {"result": {"type": "string"}},
+            "x-fastmcp-wrap-result": True,
+        }
+        assert new_tool.output_schema == expected_schema
+        assert new_tool.output_schema == base_string_tool.output_schema
+
+    def test_transform_with_explicit_output_schema_false(self, base_string_tool):
+        """Test that output_schema=False disables structured output."""
+        new_tool = Tool.from_tool(base_string_tool, output_schema=False)
+
+        assert new_tool.output_schema is None
+
+    async def test_transform_output_schema_false_runtime(self, base_string_tool):
+        """Test runtime behavior with output_schema=False."""
+        new_tool = Tool.from_tool(base_string_tool, output_schema=False)
+
+        # Debug: check that output_schema is actually None
+        assert new_tool.output_schema is None, (
+            f"Expected None, got {new_tool.output_schema}"
+        )
+
+        result = await new_tool.run({"x": 5})
+        assert result.structured_content is None
+        assert result.content[0].text == "Result: 5"  # type: ignore[attr-defined]
+
+    def test_transform_with_explicit_output_schema_dict(self, base_string_tool):
+        """Test that explicit output schema overrides parent."""
+        custom_schema = {
+            "type": "object",
+            "properties": {"message": {"type": "string"}},
+        }
+        new_tool = Tool.from_tool(base_string_tool, output_schema=custom_schema)
+
+        assert new_tool.output_schema == custom_schema
+        assert new_tool.output_schema != base_string_tool.output_schema
+
+    async def test_transform_explicit_schema_runtime(self, base_string_tool):
+        """Test runtime behavior with explicit output schema."""
+        custom_schema = {"type": "string", "minLength": 1}
+        new_tool = Tool.from_tool(base_string_tool, output_schema=custom_schema)
+
+        result = await new_tool.run({"x": 10})
+        # Non-object explicit schemas disable structured content
+        assert result.structured_content is None
+        assert result.content[0].text == "Result: 10"  # type: ignore[attr-defined]
+
+    def test_transform_with_custom_function_inferred_schema(self, base_dict_tool):
+        """Test that custom function's output schema is inferred."""
+
+        async def custom_fn(x: int) -> str:
+            result = await forward(x=x)
+            return f"Custom: {result.content[0].text}"  # type: ignore[attr-defined]
+
+        new_tool = Tool.from_tool(base_dict_tool, transform_fn=custom_fn)
+
+        # Should infer string schema from custom function and wrap it
+        expected_schema = {
+            "type": "object",
+            "properties": {"result": {"type": "string"}},
+            "x-fastmcp-wrap-result": True,
+        }
+        assert new_tool.output_schema == expected_schema
+
+    async def test_transform_custom_function_runtime(self, base_dict_tool):
+        """Test runtime behavior with custom function that has inferred schema."""
+
+        async def custom_fn(x: int) -> str:
+            result = await forward(x=x)
+            return f"Custom: {result.content[0].text}"  # type: ignore[attr-defined]
+
+        new_tool = Tool.from_tool(base_dict_tool, transform_fn=custom_fn)
+
+        result = await new_tool.run({"x": 3})
+        # Should wrap string result
+        assert result.structured_content == {"result": 'Custom: {\n  "value": 3\n}'}
+
+    def test_transform_custom_function_fallback_to_parent(self, base_string_tool):
+        """Test that custom function without output annotation falls back to parent."""
+
+        async def custom_fn(x: int):
+            # No return annotation - should fallback to parent schema
+            result = await forward(x=x)
+            return result
+
+        new_tool = Tool.from_tool(base_string_tool, transform_fn=custom_fn)
+
+        # Should use parent's schema since custom function has no annotation
+        assert new_tool.output_schema == base_string_tool.output_schema
+
+    def test_transform_custom_function_explicit_overrides(self, base_string_tool):
+        """Test that explicit output_schema overrides both custom function and parent."""
+
+        async def custom_fn(x: int) -> dict[str, str]:
+            return {"custom": "value"}
+
+        explicit_schema = {"type": "array", "items": {"type": "number"}}
+        new_tool = Tool.from_tool(
+            base_string_tool, transform_fn=custom_fn, output_schema=explicit_schema
+        )
+
+        # Explicit schema should win
+        assert new_tool.output_schema == explicit_schema
+
+    async def test_transform_custom_function_object_return(self, base_string_tool):
+        """Test custom function returning object type."""
+
+        async def custom_fn(x: int) -> dict[str, int]:
+            result = await forward(x=x)
+            return {"original": x, "transformed": x * 2}
+
+        new_tool = Tool.from_tool(base_string_tool, transform_fn=custom_fn)
+
+        # Object types should not be wrapped
+        expected_schema = TypeAdapter(dict[str, int]).json_schema()
+        assert new_tool.output_schema == expected_schema
+        assert "x-fastmcp-wrap-result" not in new_tool.output_schema
+
+        result = await new_tool.run({"x": 4})
+        # Direct value, not wrapped
+        assert result.structured_content == {"original": 4, "transformed": 8}
+
+    async def test_transform_preserves_wrap_marker_behavior(self, base_string_tool):
+        """Test that wrap marker behavior is preserved through transformation."""
+        new_tool = Tool.from_tool(base_string_tool)
+
+        result = await new_tool.run({"x": 7})
+        # Should wrap because parent schema has wrap marker
+        assert result.structured_content == {"result": "Result: 7"}
+        assert "x-fastmcp-wrap-result" in new_tool.output_schema
+
+    def test_transform_chained_output_schema_inheritance(self, base_string_tool):
+        """Test output schema inheritance through multiple transformations."""
+        # First transformation keeps parent schema
+        tool1 = Tool.from_tool(base_string_tool)
+        assert tool1.output_schema == base_string_tool.output_schema
+
+        # Second transformation also inherits
+        tool2 = Tool.from_tool(tool1)
+        assert (
+            tool2.output_schema == tool1.output_schema == base_string_tool.output_schema
+        )
+
+        # Third transformation with explicit override
+        custom_schema = {"type": "number"}
+        tool3 = Tool.from_tool(tool2, output_schema=custom_schema)
+        assert tool3.output_schema == custom_schema
+        assert tool3.output_schema != tool2.output_schema
+
+    async def test_transform_mixed_structured_unstructured_content(
+        self, base_string_tool
+    ):
+        """Test transformation handling of mixed content types."""
+
+        async def custom_fn(x: int) -> list:
+            # Return mixed content including ToolResult
+            if x == 1:
+                return ["text", {"data": x}]
+            else:
+                # Return ToolResult directly
+                return ToolResult(
+                    content=[TextContent(type="text", text=f"Custom: {x}")],
+                    structured_content={"custom_value": x},
+                )
+
+        new_tool = Tool.from_tool(base_string_tool, transform_fn=custom_fn)
+
+        # Test mixed content return
+        result1 = await new_tool.run({"x": 1})
+        assert result1.structured_content == {"result": ["text", {"data": 1}]}
+
+        # Test ToolResult return
+        result2 = await new_tool.run({"x": 2})
+        assert result2.structured_content == {"custom_value": 2}
+        assert result2.content[0].text == "Custom: 2"  # type: ignore[attr-defined]
+
+    def test_transform_output_schema_with_arg_transforms(self, base_string_tool):
+        """Test that output schema works correctly with argument transformations."""
+
+        async def custom_fn(new_x: int) -> dict[str, str]:
+            result = await forward(new_x=new_x)
+            return {"transformed": result.content[0].text}  # type: ignore[attr-defined]
+
+        new_tool = Tool.from_tool(
+            base_string_tool,
+            transform_fn=custom_fn,
+            transform_args={"x": ArgTransform(name="new_x")},
+        )
+
+        # Should infer object schema from custom function
+        expected_schema = TypeAdapter(dict[str, str]).json_schema()
+        assert new_tool.output_schema == expected_schema
+
+    async def test_transform_output_schema_none_vs_false(self, base_string_tool):
+        """Test None vs False behavior for output_schema in transforms."""
+        # None (default) should use smart fallback (inherit from parent)
+        tool_none = Tool.from_tool(base_string_tool)  # default output_schema=None
+        assert tool_none.output_schema == base_string_tool.output_schema  # Inherits
+
+        # False should explicitly disable
+        tool_false = Tool.from_tool(base_string_tool, output_schema=False)
+        assert tool_false.output_schema is None
+
+        # Different behavior at runtime
+        result_none = await tool_none.run({"x": 5})
+        result_false = await tool_false.run({"x": 5})
+
+        assert result_none.structured_content == {
+            "result": "Result: 5"
+        }  # Inherits wrapping
+        assert result_false.structured_content is None  # Disabled
+        assert result_none.content[0].text == result_false.content[0].text
+
+    async def test_transform_output_schema_with_tool_result_return(
+        self, base_string_tool
+    ):
+        """Test transform when custom function returns ToolResult directly."""
+
+        async def custom_fn(x: int) -> ToolResult:
+            # Custom function returns ToolResult - should bypass schema handling
+            return ToolResult(
+                content=[TextContent(type="text", text=f"Direct: {x}")],
+                structured_content={"direct_value": x, "doubled": x * 2},
+            )
+
+        new_tool = Tool.from_tool(base_string_tool, transform_fn=custom_fn)
+
+        # ToolResult return type should result in None output schema
+        assert new_tool.output_schema is None
+
+        result = await new_tool.run({"x": 6})
+        # Should use ToolResult content directly
+        assert result.content[0].text == "Direct: 6"  # type: ignore[attr-defined]
+        assert result.structured_content == {"direct_value": 6, "doubled": 12}
diff --git a/tests/utilities/test_json_schema_type.py b/tests/utilities/test_json_schema_type.py
index 14bf949fe..866facbce 100644
--- a/tests/utilities/test_json_schema_type.py
+++ b/tests/utilities/test_json_schema_type.py
@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Union
+from typing import Any, Union
 
 import pytest
 from pydantic import AnyUrl, BaseModel, TypeAdapter, ValidationError
@@ -326,6 +326,29 @@ class TestObjectTypes:
             }
         )
 
+    @pytest.mark.parametrize(
+        "input_type, expected_type",
+        [
+            # Plain dict becomes dict[str, Any] (JSON Schema accurate)
+            (dict, dict[str, Any]),
+            # dict[str, Any] stays the same
+            (dict[str, Any], dict[str, Any]),
+            # Simple typed dicts work correctly
+            (dict[str, str], dict[str, str]),
+            (dict[str, int], dict[str, int]),
+            # Union value types work
+            (dict[str, str | int], dict[str, str | int]),
+            # Key types are constrained to str in JSON Schema
+            (dict[int, list[str]], dict[str, list[str]]),
+            # Union key types become str (JSON Schema limitation)
+            (dict[str | int, str | None], dict[str, str | None]),
+        ],
+    )
+    def test_dict_types_are_generated_correctly(self, input_type, expected_type):
+        schema = TypeAdapter(input_type).json_schema()
+        generated_type = json_schema_to_type(schema)
+        assert generated_type == expected_type
+
     def test_object_accepts_valid(self, simple_object):
         validator = TypeAdapter(simple_object)
         result = validator.validate_python({"name": "test", "age": 30})