Support configuring immediate LRO result

Author: LucaButBoring

examples/snippets/clients/async_tools_client.py

@@ -1,5 +1,14 @@
 """
-Client example showing how to use async tools.
+Client example showing how to use async tools, including immediate result functionality.
+
+This example demonstrates:
+- Synchronous tools (immediate response)
+- Hybrid tools (sync/async modes)
+- Async-only tools (background execution with polling)
+- Batch processing with progress updates
+- Data processing pipelines
+- Elicitation (user input during async execution)
+- Immediate result tools (instant feedback + async execution)
 
 cd to the `examples/snippets` directory and run:
     uv run async-tools-client
@@ -222,6 +231,61 @@ async def demonstrate_elicitation(session: ClientSession):
             await asyncio.sleep(0.5)
 
 
+async def test_immediate_result_tool(session: ClientSession):
+    """Test calling async tool with immediate result functionality.
+
+    This demonstrates the immediate_result feature where async tools can provide
+    instant feedback while continuing to execute in the background.
+    """
+    print("\n=== Immediate Result Tool Demo ===")
+
+    # Call the async tool with immediate_result functionality
+    result = await session.call_tool("long_running_analysis", arguments={"operation": "data_processing"})
+
+    # Display immediate feedback (should be available immediately)
+    print("Immediate response received:")
+    if result.content:
+        for content in result.content:
+            if isinstance(content, types.TextContent):
+                print(f"  📋 {content.text}")
+    else:
+        print("  (No immediate content received)")
+
+    # Check if there's an async operation to poll
+    if result.operation:
+        token = result.operation.token
+        print(f"\nAsync operation started with token: {token}")
+        print("Polling for final results...")
+
+        # Poll for status updates and final result
+        while True:
+            status = await session.get_operation_status(token)
+            print(f"  Status: {status.status}")
+
+            if status.status == "completed":
+                # Get the final result
+                final_result = await session.get_operation_result(token)
+                print("\nFinal result received:")
+                for content in final_result.result.content:
+                    if isinstance(content, types.TextContent):
+                        print(f"  ✅ {content.text}")
+                break
+            elif status.status == "failed":
+                print(f"  ❌ Operation failed: {status.error}")
+                break
+            elif status.status in ("canceled", "unknown"):
+                print(f"  ⚠️ Operation ended with status: {status.status}")
+                break
+
+            # Wait before polling again
+            await asyncio.sleep(1)
+    else:
+        # This shouldn't happen for async tools, but handle gracefully
+        print("⚠️ Unexpected: tool returned synchronous result instead of async operation")
+
+    print("Immediate result demonstration complete!")
+
+
 async def run():
     """Run all async tool demonstrations."""
     # Determine protocol version from command line
@@ -261,6 +325,7 @@ async def run():
             await demonstrate_batch_processing(session)
             await demonstrate_data_processing(session)
             await demonstrate_elicitation(session)
+            await test_immediate_result_tool(session)
 
             print("\n=== All demonstrations complete! ===")
 

examples/snippets/servers/async_tools.py

@@ -9,6 +9,7 @@
 
 from pydantic import BaseModel, Field
 
+from mcp import types
 from mcp.server.fastmcp import Context, FastMCP
 
 # Create an MCP server with async operations support
@@ -206,5 +207,25 @@ async def quick_expiry_task(message: str, ctx: Context) -> str:  # type: ignore[
     return f"Quick task completed: {message} (expires in 2 seconds)"
 
 
+async def immediate_feedback(operation: str) -> list[types.ContentBlock]:
+    """Provide immediate feedback for long-running operations."""
+    return [types.TextContent(type="text", text=f"🚀 Starting {operation}... This may take a moment.")]
+
+
+@mcp.tool(invocation_modes=["async"], immediate_result=immediate_feedback)
+async def long_running_analysis(operation: str, ctx: Context) -> str:  # type: ignore[type-arg]
+    """Perform analysis with immediate user feedback."""
+    await ctx.info(f"Beginning {operation} analysis")
+
+    # Simulate long-running work with progress updates
+    for i in range(5):
+        await asyncio.sleep(1)
+        progress = (i + 1) / 5
+        await ctx.report_progress(progress, 1.0, f"Step {i + 1}/5 complete")
+
+    await ctx.info(f"Analysis '{operation}' completed successfully!")
+    return f"Analysis '{operation}' completed successfully with detailed results!"
+
+
 if __name__ == "__main__":
     mcp.run()

src/mcp/server/fastmcp/server.py

@@ -21,6 +21,7 @@
 from starlette.routing import Mount, Route
 from starlette.types import Receive, Scope, Send
 
+import mcp.types as types
 from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
 from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider, ProviderTokenVerifier, TokenVerifier
@@ -364,6 +365,10 @@ async def list_tools(self) -> list[MCPTool]:
                 annotations=info.annotations,
                 invocationMode=self._get_invocation_mode(info, client_supports_async),
                 _meta=info.meta,
+                internal=types.InternalToolProperties(
+                    immediate_result=info.immediate_result,
+                    keepalive=info.meta.get("_keep_alive") if info.meta else None,
+                ),
             )
             for info in tools
             if client_supports_async or info.invocation_modes != ["async"]
@@ -438,6 +443,7 @@ def add_tool(
         structured_output: bool | None = None,
         invocation_modes: list[InvocationMode] | None = None,
         keep_alive: int | None = None,
+        immediate_result: Callable[..., Awaitable[list[ContentBlock]]] | None = None,
     ) -> None:
         """Add a tool to the server.
 
@@ -458,6 +464,8 @@ def add_tool(
                 - If None, defaults to ["sync"] for backwards compatibility
             keep_alive: How long (in seconds) async operation results should be kept available.
                 Only applies to async tools.
+            immediate_result: Optional async function that returns immediate feedback content
+                for async tools. Must return list[ContentBlock]. Only valid for async-compatible tools.
         """
         self._tool_manager.add_tool(
             fn,
@@ -468,6 +476,7 @@ def add_tool(
             structured_output=structured_output,
             invocation_modes=invocation_modes,
             keep_alive=keep_alive,
+            immediate_result=immediate_result,
         )
 
     def tool(
@@ -479,6 +488,7 @@ def tool(
         structured_output: bool | None = None,
         invocation_modes: list[InvocationMode] | None = None,
         keep_alive: int | None = None,
+        immediate_result: Callable[..., Awaitable[list[ContentBlock]]] | None = None,
     ) -> Callable[[AnyFunction], AnyFunction]:
         """Decorator to register a tool.
 
@@ -501,6 +511,8 @@ def tool(
                 - Tools with "async" mode will be hidden from clients that don't support async execution
             keep_alive: How long (in seconds) async operation results should be kept available.
                 Only applies to async tools.
+            immediate_result: Optional async function that returns immediate feedback content
+                for async tools. Must return list[ContentBlock]. Only valid for async-compatible tools.
 
         Example:
             @server.tool()
@@ -527,6 +539,15 @@ async def async_only_tool(data: str, ctx: Context) -> str:
             def hybrid_tool(x: int) -> str:
                 # This tool supports both sync and async execution
                 return str(x)
+
+            async def immediate_feedback(operation: str) -> list[ContentBlock]:
+                return [TextContent(type="text", text=f"Starting {operation}...")]
+
+            @server.tool(invocation_modes=["async"], immediate_result=immediate_feedback)
+            async def long_running_tool(operation: str, ctx: Context) -> str:
+                # This tool provides immediate feedback while running asynchronously
+                await ctx.info(f"Processing {operation}")
+                return f"Completed {operation}"
         """
         # Check if user passed function directly instead of calling decorator
         if callable(name):
@@ -544,6 +565,7 @@ def decorator(fn: AnyFunction) -> AnyFunction:
                 structured_output=structured_output,
                 invocation_modes=invocation_modes,
                 keep_alive=keep_alive,
+                immediate_result=immediate_result,
             )
             return fn
 

src/mcp/server/fastmcp/tools/base.py

@@ -11,7 +11,7 @@
 from mcp.server.fastmcp.exceptions import ToolError
 from mcp.server.fastmcp.utilities.context_injection import find_context_parameter
 from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata, func_metadata
-from mcp.types import ToolAnnotations
+from mcp.types import ContentBlock, ToolAnnotations
 
 if TYPE_CHECKING:
     from mcp.server.fastmcp.server import Context
@@ -38,10 +38,10 @@ class Tool(BaseModel):
     invocation_modes: list[InvocationMode] = Field(
         default=["sync"], description="Supported invocation modes (sync/async)"
     )
-    meta: dict[str, Any] | None = Field(description="Optional additional tool information.", default=None)
-    immediate_result: Callable[..., Awaitable[list[Any]]] | None = Field(
+    immediate_result: Callable[..., Awaitable[list[ContentBlock]]] | None = Field(
         None, exclude=True, description="Optional immediate result function for async tools"
     )
+    meta: dict[str, Any] | None = Field(description="Optional additional tool information.", default=None)
 
     @cached_property
     def output_schema(self) -> dict[str, Any] | None:
@@ -59,8 +59,8 @@ def from_function(
         structured_output: bool | None = None,
         invocation_modes: list[InvocationMode] | None = None,
         keep_alive: int | None = None,
-        meta: dict[str, Any] | None = None,
         immediate_result: Callable[..., Awaitable[list[Any]]] | None = None,
+        meta: dict[str, Any] | None = None,
     ) -> Tool:
         """Create a Tool from a function."""
         func_name = name or fn.__name__
@@ -129,8 +129,8 @@ def from_function(
             context_kwarg=context_kwarg,
             annotations=annotations,
             invocation_modes=invocation_modes,
-            meta=meta,
             immediate_result=immediate_result,
+            meta=meta,
         )
 
     async def run(

src/mcp/server/fastmcp/tools/tool_manager.py

@@ -1,13 +1,13 @@
 from __future__ import annotations as _annotations
 
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from typing import TYPE_CHECKING, Any
 
 from mcp.server.fastmcp.exceptions import ToolError
 from mcp.server.fastmcp.tools.base import InvocationMode, Tool
 from mcp.server.fastmcp.utilities.logging import get_logger
 from mcp.shared.context import LifespanContextT, RequestT
-from mcp.types import ToolAnnotations
+from mcp.types import ContentBlock, ToolAnnotations
 
 if TYPE_CHECKING:
     from mcp.server.fastmcp.server import Context
@@ -52,6 +52,7 @@ def add_tool(
         structured_output: bool | None = None,
         invocation_modes: list[InvocationMode] | None = None,
         keep_alive: int | None = None,
+        immediate_result: Callable[..., Awaitable[list[ContentBlock]]] | None = None,
         meta: dict[str, Any] | None = None,
     ) -> Tool:
         """Add a tool to the server."""
@@ -64,6 +65,7 @@ def add_tool(
             structured_output=structured_output,
             invocation_modes=invocation_modes,
             keep_alive=keep_alive,
+            immediate_result=immediate_result,
             meta=meta,
         )
         existing = self._tools.get(tool.name)

src/mcp/server/lowlevel/server.py

@@ -469,6 +469,23 @@ async def handler(req: types.CallToolRequest):
                     # Check for async execution
                     if tool and self.async_operations and self._should_execute_async(tool):
                         keep_alive = self._get_tool_keep_alive(tool)
+                        immediate_content: list[types.ContentBlock] = []
+
+                        # Execute immediate result if available
+                        if self._has_immediate_result(tool):
+                            try:
+                                immediate_content = await self._execute_immediate_result(tool, arguments)
+                                logger.debug(f"Executed immediate result for {tool_name}")
+                            except McpError:
+                                # Re-raise McpError as-is
+                                raise
+                            except Exception as e:
+                                raise McpError(
+                                    types.ErrorData(
+                                        code=types.INTERNAL_ERROR,
+                                        message=f"Immediate result execution failed: {str(e)}",
+                                    )
+                                )
 
                         # Create async operation
                         operation = self.async_operations.create_operation(
@@ -499,11 +516,11 @@ async def execute_async():
 
                         asyncio.create_task(execute_async())
 
-                        # Return operation result immediately
+                        # Return operation result with immediate content
                         logger.info(f"Returning async operation result for {tool_name}")
                         return types.ServerResult(
                             types.CallToolResult(
-                                content=[],
+                                content=immediate_content,
                                 operation=types.AsyncResultProperties(
                                     token=operation.token,
                                     keepAlive=operation.keep_alive,
@@ -588,9 +605,34 @@ def _should_execute_async(self, tool: types.Tool) -> bool:
 
     def _get_tool_keep_alive(self, tool: types.Tool) -> int:
         """Get the keepalive value for an async tool."""
-        if not tool.meta or "_keep_alive" not in tool.meta:
-            raise ValueError(f"_keep_alive not defined for tool {tool.name}")
-        return cast(int, tool.meta["_keep_alive"])
+        if tool.internal.keepalive is None:
+            raise ValueError(f"keepalive not defined for tool {tool.name}")
+        return tool.internal.keepalive
+
+    def _has_immediate_result(self, tool: types.Tool) -> bool:
+        """Check if tool has immediate_result function."""
+        return tool.internal.immediate_result is not None and callable(tool.internal.immediate_result)
+
+    async def _execute_immediate_result(self, tool: types.Tool, arguments: dict[str, Any]) -> list[types.ContentBlock]:
+        """Execute immediate result function and return content blocks."""
+        immediate_fn = tool.internal.immediate_result
+
+        if immediate_fn is None:
+            raise ValueError(f"No immediate_result function found for tool {tool.name}")
+
+        # Validate function signature and execute
+        try:
+            result = await immediate_fn(**arguments)
+            if not isinstance(result, list):
+                raise ValueError("immediate_result must return list[ContentBlock]")
+            return cast(list[types.ContentBlock], result)
+        except McpError:
+            # Re-raise McpError as-is
+            raise
+        except Exception as e:
+            raise McpError(
+                types.ErrorData(code=types.INTERNAL_ERROR, message=f"Immediate result execution error: {str(e)}")
+            )
 
     def progress_notification(self):
         def decorator(

src/mcp/types.py

@@ -858,6 +858,18 @@ class ToolAnnotations(BaseModel):
     model_config = ConfigDict(extra="allow")
 
 
+class InternalToolProperties(BaseModel):
+    """
+    Internal properties for tools that are not serialized in the MCP protocol.
+    """
+
+    immediate_result: Any = Field(default=None)
+    """Function to execute for immediate results in async operations."""
+
+    keepalive: int | None = Field(default=None)
+    """Keepalive duration in seconds for async operations."""
+
+
 class Tool(BaseMetadata):
     """Definition for a tool the client can call."""
 
@@ -883,6 +895,10 @@ class Tool(BaseMetadata):
     See [MCP specification](https://github.com/modelcontextprotocol/modelcontextprotocol/blob/47339c03c143bb4ec01a26e721a1b8fe66634ebe/docs/specification/draft/basic/index.mdx#general-fields)
     for notes on _meta usage.
     """
+    internal: InternalToolProperties = Field(default_factory=InternalToolProperties, exclude=True)
+    """
+    Internal properties not serialized in MCP protocol.
+    """
     model_config = ConfigDict(extra="allow")