Implement lowlevel async CallTool

Author: LucaButBoring

src/mcp/server/lowlevel/server.py

@@ -67,6 +67,7 @@ async def main():
 
 from __future__ import annotations as _annotations
 
+import asyncio
 import contextvars
 import json
 import logging
@@ -465,46 +466,55 @@ async def handler(req: types.CallToolRequest):
                         except jsonschema.ValidationError as e:
                             return self._make_error_result(f"Input validation error: {e.message}")
 
-                    # tool call
-                    results = await func(tool_name, arguments)
+                    # Check for async execution
+                    if tool and self.async_operations and self._should_execute_async(tool):
+                        # Create async operation
+                        session_id = f"session_{id(self.request_context.session)}"
+                        operation = self.async_operations.create_operation(
+                            tool_name=tool_name,
+                            arguments=arguments,
+                            session_id=session_id,
+                        )
+                        logger.debug(f"Created async operation with token: {operation.token}")
 
-                    # output normalization
-                    unstructured_content: UnstructuredContent
-                    maybe_structured_content: StructuredContent | None
-                    if 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):
-                        # tool returned structured content only
-                        maybe_structured_content = cast(StructuredContent, results)
-                        unstructured_content = [types.TextContent(type="text", text=json.dumps(results, indent=2))]
-                    elif hasattr(results, "__iter__"):
-                        # tool returned unstructured content only
-                        unstructured_content = cast(UnstructuredContent, results)
-                        maybe_structured_content = None
-                    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:
-                            return self._make_error_result(
-                                "Output validation error: outputSchema defined but no structured output returned"
-                            )
-                        else:
+                        # Start async execution in background
+                        async def execute_async():
                             try:
-                                jsonschema.validate(instance=maybe_structured_content, 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,
+                                logger.debug(f"Starting async execution of {tool_name}")
+                                results = await func(tool_name, arguments)
+                                logger.debug(f"Async execution completed for {tool_name}")
+
+                                # Process results using shared logic
+                                result = self._process_tool_result(results, tool)
+                                self.async_operations.complete_operation(operation.token, result)
+                                logger.debug(f"Completed async operation {operation.token}")
+                            except Exception as e:
+                                logger.exception(f"Async execution failed for {tool_name}")
+                                self.async_operations.fail_operation(operation.token, str(e))
+
+                        asyncio.create_task(execute_async())
+
+                        # Return operation result immediately
+                        logger.info(f"Returning async operation result for {tool_name}")
+                        return types.ServerResult(
+                            types.CallToolResult(
+                                content=[],
+                                operation=types.AsyncResultProperties(
+                                    token=operation.token,
+                                    keepAlive=3600,
+                                ),
+                            )
                         )
-                    )
+
+                    # tool call
+                    results = await func(tool_name, arguments)
+
+                    # Process results using shared logic
+                    try:
+                        result = self._process_tool_result(results, tool)
+                        return types.ServerResult(result)
+                    except ValueError as e:
+                        return self._make_error_result(str(e))
                 except Exception as e:
                     return self._make_error_result(str(e))
 
@@ -513,6 +523,61 @@ async def handler(req: types.CallToolRequest):
 
         return decorator
 
+    def _process_tool_result(
+        self, results: UnstructuredContent | StructuredContent | CombinationContent, tool: types.Tool | None = None
+    ) -> types.CallToolResult:
+        """Process tool results and create CallToolResult with validation."""
+        # output normalization
+        unstructured_content: UnstructuredContent
+        maybe_structured_content: StructuredContent | None
+        if 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):
+            # tool returned structured content only
+            maybe_structured_content = cast(StructuredContent, results)
+            unstructured_content = [types.TextContent(type="text", text=json.dumps(results, indent=2))]
+        elif hasattr(results, "__iter__"):
+            # tool returned unstructured content only
+            unstructured_content = cast(UnstructuredContent, results)
+            maybe_structured_content = None
+        else:
+            raise ValueError(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:
+                raise ValueError("Output validation error: outputSchema defined but no structured output returned")
+            else:
+                try:
+                    jsonschema.validate(instance=maybe_structured_content, schema=tool.outputSchema)
+                except jsonschema.ValidationError as e:
+                    raise ValueError(f"Output validation error: {e.message}")
+
+        # result
+        return types.CallToolResult(
+            content=list(unstructured_content),
+            structuredContent=maybe_structured_content,
+            isError=False,
+        )
+
+    def _should_execute_async(self, tool: types.Tool) -> bool:
+        """Check if a tool should be executed asynchronously."""
+        # Check if client supports async tools (protocol version "next")
+        try:
+            if self.request_context and self.request_context.session.client_params:
+                client_version = str(self.request_context.session.client_params.protocolVersion)
+                if client_version != "next":
+                    return False
+            else:
+                return False
+        except (AttributeError, ValueError):
+            return False
+
+        # Check if tool is async-only
+        invocation_mode = getattr(tool, "invocationMode", None)
+        return invocation_mode == "async"
+
     def progress_notification(self):
         def decorator(
             func: Callable[[str | int, float, float | None, str | None], Awaitable[None]],
@@ -783,9 +848,9 @@ async def _handle_request(
                 # Track async operations for cancellation
                 if isinstance(req, types.CallToolRequest):
                     result = response.root
-                    if isinstance(result, types.CallToolResult) and result.operation_result is not None:
+                    if isinstance(result, types.CallToolResult) and result.operation is not None:
                         # This is an async operation, track the request ID to token mapping
-                        operation_token = result.operation_result.token
+                        operation_token = result.operation.token
                         self._request_to_operation[message.request_id] = operation_token
                         logger.debug(f"Tracking async operation {operation_token} for request {message.request_id}")
 

src/mcp/types.py

@@ -130,7 +130,7 @@ class Operation(BaseModel):
     See [MCP specification](https://github.com/modelcontextprotocol/modelcontextprotocol/blob/47339c03c143bb4ec01a26e721a1b8fe66634ebe/docs/specification/draft/basic/index.mdx#general-fields)
     for notes on _meta usage.
     """
-    operation: Operation | None = Field(alias="_operation", default=None)
+    _operation: Operation | None = None
     """
     Async operation parameters, only used when a result is sent in response to a request with operation parameters.
     """
@@ -992,7 +992,7 @@ class CallToolResult(Result):
     structuredContent: dict[str, Any] | None = None
     """An optional JSON object that represents the structured result of the tool call."""
     isError: bool = False
-    operation_result: AsyncResultProperties | None = Field(serialization_alias="operation", default=None)
+    operation: AsyncResultProperties | None = Field(default=None)
     """Optional async execution information. Present when tool is executed asynchronously."""
 
 

tests/server/fastmcp/test_server.py

@@ -1,3 +1,4 @@
+import asyncio
 import base64
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
@@ -651,6 +652,103 @@ def hybrid_tool(x: int) -> int:
             # Hybrid tools should not have invocationMode field (None) for old clients
             assert tool.invocationMode is None
 
+    @pytest.mark.anyio
+    async def test_async_tool_call_basic(self):
+        """Test basic async tool call functionality."""
+        mcp = FastMCP("AsyncTest")
+
+        @mcp.tool(invocation_modes=["async"])
+        async def async_add(a: int, b: int) -> int:
+            """Add two numbers asynchronously."""
+            await asyncio.sleep(0.01)  # Simulate async work
+            return a + b
+
+        async with client_session(mcp._mcp_server, protocol_version="next") as client:
+            result = await client.call_tool("async_add", {"a": 5, "b": 3})
+
+            # Should get operation token for async call
+            assert result.operation is not None
+            token = result.operation.token
+
+            # Poll for completion
+            while True:
+                status = await client.get_operation_status(token)
+                if status.status == "completed":
+                    final_result = await client.get_operation_result(token)
+                    assert not final_result.result.isError
+                    assert len(final_result.result.content) == 1
+                    content = final_result.result.content[0]
+                    assert isinstance(content, TextContent)
+                    assert content.text == "8"
+                    break
+                elif status.status == "failed":
+                    pytest.fail(f"Operation failed: {status.error}")
+                await asyncio.sleep(0.01)
+
+    @pytest.mark.anyio
+    async def test_async_tool_call_structured_output(self):
+        """Test async tool call with structured output."""
+        mcp = FastMCP("AsyncTest")
+
+        class AsyncResult(BaseModel):
+            value: int
+            processed: bool = True
+
+        @mcp.tool(invocation_modes=["async"])
+        async def async_structured_tool(x: int) -> AsyncResult:
+            """Process data and return structured result."""
+            await asyncio.sleep(0.01)  # Simulate async work
+            return AsyncResult(value=x * 2)
+
+        async with client_session(mcp._mcp_server, protocol_version="next") as client:
+            result = await client.call_tool("async_structured_tool", {"x": 21})
+
+            # Should get operation token for async call
+            assert result.operation is not None
+            token = result.operation.token
+
+            # Poll for completion
+            while True:
+                status = await client.get_operation_status(token)
+                if status.status == "completed":
+                    final_result = await client.get_operation_result(token)
+                    assert not final_result.result.isError
+                    assert final_result.result.structuredContent is not None
+                    assert final_result.result.structuredContent == {"value": 42, "processed": True}
+                    break
+                elif status.status == "failed":
+                    pytest.fail(f"Operation failed: {status.error}")
+                await asyncio.sleep(0.01)
+
+    @pytest.mark.anyio
+    async def test_async_tool_call_validation_error(self):
+        """Test async tool call with server-side validation error."""
+        mcp = FastMCP("AsyncTest")
+
+        @mcp.tool(invocation_modes=["async"])
+        async def async_invalid_tool() -> list[int]:
+            """Tool that returns invalid structured output."""
+            await asyncio.sleep(0.01)  # Simulate async work
+            return [1, 2, 3, [4]]  # type: ignore
+
+        async with client_session(mcp._mcp_server, protocol_version="next") as client:
+            result = await client.call_tool("async_invalid_tool", {})
+
+            # Should get operation token for async call
+            assert result.operation is not None
+            token = result.operation.token
+
+            # Poll for completion - should fail due to validation error
+            while True:
+                status = await client.get_operation_status(token)
+                if status.status == "failed":
+                    # Operation should fail due to validation error
+                    assert status.error is not None
+                    break
+                elif status.status == "completed":
+                    pytest.fail("Operation should have failed due to validation error")
+                await asyncio.sleep(0.01)
+
 
 class TestServerResources:
     @pytest.mark.anyio