Allow CallToolResult to be returned directly from tool calls

Author: BrandonShar

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")
@@ -502,42 +502,46 @@ async def handler(req: types.CallToolRequest):
                     results = await func(tool_name, arguments)
 
                     # output normalization
+                    casted_results: types.CallToolResult
                     unstructured_content: UnstructuredContent
                     maybe_structured_content: StructuredContent | None
-                    if isinstance(results, tuple) and len(results) == 2:
+                    if isinstance(results, types.CallToolResult):
+                        casted_results = results
+                    elif isinstance(results, tuple) and len(results) == 2:
                         # tool returned both structured and unstructured content
                         unstructured_content, maybe_structured_content = cast(CombinationContent, results)
+                        casted_results = types.CallToolResult(
+                            content=list(unstructured_content),
+                            structuredContent=maybe_structured_content,
+                        )
                     elif isinstance(results, dict):
                         # tool returned structured content only
-                        maybe_structured_content = cast(StructuredContent, results)
-                        unstructured_content = [types.TextContent(type="text", text=json.dumps(results, indent=2))]
+                        casted_results = types.CallToolResult(
+                            content=[types.TextContent(type="text", text=json.dumps(results, indent=2))],
+                            structuredContent=cast(StructuredContent, results),
+                        )
                     elif hasattr(results, "__iter__"):
                         # tool returned unstructured content only
-                        unstructured_content = cast(UnstructuredContent, results)
-                        maybe_structured_content = None
+                        casted_results = types.CallToolResult(
+                            content=list(cast(UnstructuredContent, results)),
+                        )
                     else:
                         return self._make_error_result(f"Unexpected return type from tool: {type(results).__name__}")
 
                     # output validation
                     if tool and tool.outputSchema is not None:
-                        if maybe_structured_content is None:
+                        if casted_results.structuredContent is None:
                             return self._make_error_result(
                                 "Output validation error: outputSchema defined but no structured output returned"
                             )
                         else:
                             try:
-                                jsonschema.validate(instance=maybe_structured_content, schema=tool.outputSchema)
+                                jsonschema.validate(instance=casted_results.structuredContent, schema=tool.outputSchema)
                             except jsonschema.ValidationError as e:
                                 return self._make_error_result(f"Output validation error: {e.message}")
 
                     # result
-                    return types.ServerResult(
-                        types.CallToolResult(
-                            content=list(unstructured_content),
-                            structuredContent=maybe_structured_content,
-                            isError=False,
-                        )
-                    )
+                    return types.ServerResult(casted_results)
                 except Exception as e:
                     return self._make_error_result(str(e))
 

tests/server/test_lowlevel_output_validation.py

@@ -391,6 +391,141 @@ async def test_callback(client_session: ClientSession) -> CallToolResult:
     assert result.structuredContent == {"sentiment": "positive", "confidence": 0.95}
 
 
+@pytest.mark.anyio
+async def test_tool_call_result_without_output_schema():
+    """Test returning ToolCallResult when no outputSchema is defined."""
+    tools = [
+        Tool(
+            name="get_info",
+            description="Get structured information",
+            inputSchema={
+                "type": "object",
+                "properties": {},
+            },
+            # No outputSchema defined
+        )
+    ]
+
+    async def call_tool_handler(name: str, arguments: dict[str, Any]) -> CallToolResult:
+        if name == "get_info":
+            return CallToolResult(
+                content=[TextContent(type="text", text="Results calculated")],
+                structuredContent={"status": "ok", "data": {"value": 42}},
+                _meta={"some": "metadata"},
+            )
+        else:
+            raise ValueError(f"Unknown tool: {name}")
+
+    async def test_callback(client_session: ClientSession) -> CallToolResult:
+        return await client_session.call_tool("get_info", {})
+
+    result = await run_tool_test(tools, call_tool_handler, test_callback)
+
+    # Verify results
+    assert result is not None
+    assert not result.isError
+    assert len(result.content) == 1
+    assert result.content[0].type == "text"
+    assert result.content[0].text == "Results calculated"
+    assert isinstance(result.content[0], TextContent)
+    assert result.structuredContent == {"status": "ok", "data": {"value": 42}}
+    assert result.meta == {"some": "metadata"}
+
+
+@pytest.mark.anyio
+async def test_valid_tool_call_result_with_output_schema():
+    """Test returning ToolCallResult when no outputSchema is defined."""
+    tools = [
+        Tool(
+            name="get_info",
+            description="Get structured information",
+            inputSchema={
+                "type": "object",
+                "properties": {},
+            },
+            outputSchema={
+                "type": "object",
+                "properties": {
+                    "name": {"type": "string"},
+                    "number": {"type": "integer"},
+                },
+                "required": ["name", "number"],
+            },
+        )
+    ]
+
+    async def call_tool_handler(name: str, arguments: dict[str, Any]) -> CallToolResult:
+        if name == "get_info":
+            return CallToolResult(
+                content=[TextContent(type="text", text="Results calculated")],
+                structuredContent={"name": "Brandon", "number": 29},
+                _meta={"some": "metadata"},
+            )
+        else:
+            raise ValueError(f"Unknown tool: {name}")
+
+    async def test_callback(client_session: ClientSession) -> CallToolResult:
+        return await client_session.call_tool("get_info", {})
+
+    result = await run_tool_test(tools, call_tool_handler, test_callback)
+
+    # Verify results
+    assert result is not None
+    assert not result.isError
+    assert len(result.content) == 1
+    assert result.content[0].type == "text"
+    assert result.content[0].text == "Results calculated"
+    assert isinstance(result.content[0], TextContent)
+    assert result.structuredContent == {"name": "Brandon", "number": 29}
+    assert result.meta == {"some": "metadata"}
+
+
+@pytest.mark.anyio
+async def test_invalid_tool_call_result_with_output_schema():
+    """Test returning ToolCallResult when no outputSchema is defined."""
+    tools = [
+        Tool(
+            name="get_info",
+            description="Get structured information",
+            inputSchema={
+                "type": "object",
+                "properties": {},
+            },
+            outputSchema={
+                "type": "object",
+                "properties": {
+                    "name": {"type": "string"},
+                    "number": {"type": "integer"},
+                },
+                "required": ["name", "number"],
+            },
+        )
+    ]
+
+    async def call_tool_handler(name: str, arguments: dict[str, Any]) -> CallToolResult:
+        if name == "get_info":
+            return CallToolResult(
+                content=[TextContent(type="text", text="Results calculated")],
+                structuredContent={"name": "Brandon"},
+                _meta={"some": "metadata"},
+            )
+        else:
+            raise ValueError(f"Unknown tool: {name}")
+
+    async def test_callback(client_session: ClientSession) -> CallToolResult:
+        return await client_session.call_tool("get_info", {})
+
+    result = await run_tool_test(tools, call_tool_handler, test_callback)
+
+    assert result is not None
+    assert result.isError
+    assert len(result.content) == 1
+    assert result.content[0].type == "text"
+    assert isinstance(result.content[0], TextContent)
+    assert "Output validation error:" in result.content[0].text
+    assert "'number' is a required property" in result.content[0].text
+
+
 @pytest.mark.anyio
 async def test_output_schema_type_validation():
     """Test outputSchema validates types correctly."""