Add basic documentation for async tools

Author: LucaButBoring

README.md

@@ -487,6 +487,290 @@ def get_temperature(city: str) -> float:
 _Full example: [examples/snippets/servers/structured_output.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/structured_output.py)_
 <!-- /snippet-source -->
 
+#### Async Tools
+
+Tools can be configured to run asynchronously, allowing for long-running operations that execute in the background while clients poll for status and results. Async tools currently require protocol version `next` and support operation tokens for tracking execution state.
+
+Tools can specify their invocation mode: `sync` (default), `async`, or `["sync", "async"]` for hybrid tools that support both patterns. Async tools can provide immediate feedback while continuing to execute, and support configurable keep-alive duration for result availability.
+
+<!-- snippet-source examples/snippets/servers/async_tools.py -->
+```python
+"""
+FastMCP async tools example showing different invocation modes.
+
+cd to the `examples/snippets/clients` directory and run:
+    uv run server async_tools stdio
+"""
+
+import asyncio
+
+from pydantic import BaseModel, Field
+
+from mcp import types
+from mcp.server.fastmcp import Context, FastMCP
+
+# Create an MCP server with async operations support
+mcp = FastMCP("Async Tools Demo")
+
+
+class UserPreferences(BaseModel):
+    """Schema for collecting user preferences."""
+
+    continue_processing: bool = Field(description="Should we continue with the operation?")
+    priority_level: str = Field(
+        default="normal",
+        description="Priority level: low, normal, high",
+    )
+
+
+@mcp.tool(invocation_modes=["async"])
+async def async_elicitation_tool(operation: str, ctx: Context) -> str:  # type: ignore[type-arg]
+    """An async tool that uses elicitation to get user input."""
+    await ctx.info(f"Starting operation: {operation}")
+
+    # Simulate some initial processing
+    await asyncio.sleep(0.5)
+    await ctx.report_progress(0.3, 1.0, "Initial processing complete")
+
+    # Ask user for preferences
+    result = await ctx.elicit(
+        message=f"Operation '{operation}' requires user input. How should we proceed?",
+        schema=UserPreferences,
+    )
+
+    if result.action == "accept" and result.data:
+        if result.data.continue_processing:
+            await ctx.info(f"Continuing with {result.data.priority_level} priority")
+            # Simulate processing based on user choice
+            processing_time = {"low": 0.5, "normal": 1.0, "high": 1.5}.get(result.data.priority_level, 1.0)
+            await asyncio.sleep(processing_time)
+            await ctx.report_progress(1.0, 1.0, "Operation complete")
+            return f"Operation '{operation}' completed successfully with {result.data.priority_level} priority"
+        else:
+            await ctx.warning("User chose not to continue")
+            return f"Operation '{operation}' cancelled by user"
+    else:
+        await ctx.error("User declined or cancelled the operation")
+        return f"Operation '{operation}' aborted"
+
+
+@mcp.tool()
+def sync_tool(x: int) -> str:
+    """An implicitly-synchronous tool."""
+    return f"Sync result: {x * 2}"
+
+
+@mcp.tool(invocation_modes=["async"])
+async def async_only_tool(data: str, ctx: Context) -> str:  # type: ignore[type-arg]
+    """An async-only tool that takes time to complete."""
+    await ctx.info("Starting long-running analysis...")
+
+    # Simulate long-running work with progress updates
+    for i in range(5):
+        await asyncio.sleep(0.5)
+        progress = (i + 1) / 5
+        await ctx.report_progress(progress, 1.0, f"Processing step {i + 1}/5")
+
+    await ctx.info("Analysis complete!")
+    return f"Async analysis result for: {data}"
+
+
+@mcp.tool(invocation_modes=["sync", "async"])
+def hybrid_tool(message: str, ctx: Context | None = None) -> str:  # type: ignore[type-arg]
+    """A hybrid tool that works both sync and async."""
+    if ctx:
+        # Async mode - we have context for progress reporting
+        import asyncio
+
+        async def async_work():
+            await ctx.info(f"Processing '{message}' asynchronously...")
+            await asyncio.sleep(0.5)  # Simulate some work
+            await ctx.debug("Async processing complete")
+
+        # Run the async work (this is a bit of a hack for demo purposes)
+        try:
+            loop = asyncio.get_event_loop()
+            loop.create_task(async_work())
+        except RuntimeError:
+            pass  # No event loop running
+
+    # Both sync and async modes return the same result
+    return f"Hybrid result: {message.upper()}"
+
+
+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!"
+
+
+@mcp.tool(invocation_modes=["async"], keep_alive=1800)
+async def long_running_task(task_name: str, ctx: Context) -> str:  # type: ignore[type-arg]
+    """A long-running task with custom keep_alive duration."""
+    await ctx.info(f"Starting long-running task: {task_name}")
+
+    # Simulate extended processing
+    await asyncio.sleep(2)
+    await ctx.report_progress(0.5, 1.0, "Halfway through processing")
+    await asyncio.sleep(2)
+
+    await ctx.info(f"Task '{task_name}' completed successfully")
+    return f"Long-running task '{task_name}' finished with 30-minute keep_alive"
+
+
+if __name__ == "__main__":
+    mcp.run()
+```
+
+_Full example: [examples/snippets/servers/async_tools.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/async_tools.py)_
+<!-- /snippet-source -->
+
+Clients using protocol version `next` can interact with async tools by polling operation status and retrieving results:
+
+<!-- snippet-source examples/snippets/clients/async_tools_client.py -->
+```python
+"""
+Client example showing how to use async tools, including immediate result functionality.
+
+cd to the `examples/snippets` directory and run:
+    uv run async-tools-client
+    uv run async-tools-client --protocol=latest  # backwards compatible mode
+    uv run async-tools-client --protocol=next    # async tools mode
+"""
+
+import asyncio
+import os
+import sys
+
+from mcp import ClientSession, StdioServerParameters, types
+from mcp.client.stdio import stdio_client
+from mcp.shared.context import RequestContext
+
+# Create server parameters for stdio connection
+server_params = StdioServerParameters(
+    command="uv",  # Using uv to run the server
+    args=["run", "server", "async_tools", "stdio"],
+    env={"UV_INDEX": os.environ.get("UV_INDEX", "")},
+)
+
+
+async def demonstrate_async_tool(session: ClientSession):
+    """Demonstrate calling an async-only tool."""
+    print("\n=== Asynchronous Tool Demo ===")
+
+    # Call the async tool
+    result = await session.call_tool("async_only_tool", arguments={"data": "sample dataset"})
+
+    if result.operation:
+        token = result.operation.token
+        print(f"Async operation started with token: {token}")
+
+        # Poll for status updates
+        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)
+                for content in final_result.result.content:
+                    if isinstance(content, types.TextContent):
+                        print(f"Final result: {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)
+
+
+async def test_immediate_result_tool(session: ClientSession):
+    """Test calling async tool with immediate result functionality."""
+    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}")
+
+    # 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
+
+            # Wait before polling again
+            await asyncio.sleep(1)
+
+
+async def run():
+    """Run async tool demonstrations."""
+    protocol_version = "next"  # Required for async tools support
+
+    async with stdio_client(server_params) as (read, write):
+        async with ClientSession(read, write, protocol_version=protocol_version) as session:
+            await session.initialize()
+
+            # List available tools to see invocation modes
+            tools = await session.list_tools()
+            print("Available tools:")
+            for tool in tools.tools:
+                invocation_mode = getattr(tool, "invocationMode", "sync")
+                print(f"  - {tool.name}: {tool.description} (mode: {invocation_mode})")
+
+            await demonstrate_async_tool(session)
+            await test_immediate_result_tool(session)
+
+
+if __name__ == "__main__":
+    asyncio.run(run())
+```
+
+_Full example: [examples/snippets/clients/async_tools_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/async_tools_client.py)_
+<!-- /snippet-source -->
+
+The `@mcp.tool()` decorator accepts `invocation_modes` to specify supported execution patterns, `immediate_result` to provide instant feedback for async tools, and `keep_alive` to set how long operation results remain available (default: 300 seconds).
+
 ### Prompts
 
 Prompts are reusable templates that help LLMs interact with your server effectively: