Refactor example servers to have simpler tool dispatch

Extract tool-specific logic into separate handler functions, keeping
the call_tool decorator handler simple - it just dispatches based on
tool name and returns an error for unknown tools.

Author: maxisbey

examples/servers/simple-task-interactive/mcp_simple_task_interactive/server.py

@@ -46,69 +46,78 @@ async def list_tools() -> list[types.Tool]:
     ]
 
 
-@server.call_tool()
-async def handle_call_tool(name: str, arguments: dict[str, Any]) -> types.CallToolResult | types.CreateTaskResult:
+async def handle_confirm_delete(arguments: dict[str, Any]) -> types.CreateTaskResult:
+    """Handle the confirm_delete tool - demonstrates elicitation."""
     ctx = server.request_context
-
-    # Validate task mode - this tool requires task augmentation
     ctx.experimental.validate_task_mode(types.TASK_REQUIRED)
 
-    if name == "confirm_delete":
-        filename = arguments.get("filename", "unknown.txt")
-        print(f"\n[Server] confirm_delete called for '{filename}'")
+    filename = arguments.get("filename", "unknown.txt")
+    print(f"\n[Server] confirm_delete called for '{filename}'")
 
-        async def do_confirm(task: ServerTaskContext) -> types.CallToolResult:
-            print(f"[Server] Task {task.task_id} starting elicitation...")
+    async def work(task: ServerTaskContext) -> types.CallToolResult:
+        print(f"[Server] Task {task.task_id} starting elicitation...")
 
-            result = await task.elicit(
-                message=f"Are you sure you want to delete '{filename}'?",
-                requestedSchema={
-                    "type": "object",
-                    "properties": {"confirm": {"type": "boolean"}},
-                    "required": ["confirm"],
-                },
-            )
+        result = await task.elicit(
+            message=f"Are you sure you want to delete '{filename}'?",
+            requestedSchema={
+                "type": "object",
+                "properties": {"confirm": {"type": "boolean"}},
+                "required": ["confirm"],
+            },
+        )
 
-            print(f"[Server] Received elicitation response: action={result.action}, content={result.content}")
+        print(f"[Server] Received elicitation response: action={result.action}, content={result.content}")
 
-            if result.action == "accept" and result.content:
-                confirmed = result.content.get("confirm", False)
-                text = f"Deleted '{filename}'" if confirmed else "Deletion cancelled"
-            else:
-                text = "Deletion cancelled"
+        if result.action == "accept" and result.content:
+            confirmed = result.content.get("confirm", False)
+            text = f"Deleted '{filename}'" if confirmed else "Deletion cancelled"
+        else:
+            text = "Deletion cancelled"
 
-            print(f"[Server] Completing task with result: {text}")
-            return types.CallToolResult(content=[types.TextContent(type="text", text=text)])
+        print(f"[Server] Completing task with result: {text}")
+        return types.CallToolResult(content=[types.TextContent(type="text", text=text)])
 
-        # run_task creates the task, spawns work, returns CreateTaskResult immediately
-        return await ctx.experimental.run_task(do_confirm)
+    return await ctx.experimental.run_task(work)
 
-    elif name == "write_haiku":
-        topic = arguments.get("topic", "nature")
-        print(f"\n[Server] write_haiku called for topic '{topic}'")
 
-        async def do_haiku(task: ServerTaskContext) -> types.CallToolResult:
-            print(f"[Server] Task {task.task_id} starting sampling...")
+async def handle_write_haiku(arguments: dict[str, Any]) -> types.CreateTaskResult:
+    """Handle the write_haiku tool - demonstrates sampling."""
+    ctx = server.request_context
+    ctx.experimental.validate_task_mode(types.TASK_REQUIRED)
+
+    topic = arguments.get("topic", "nature")
+    print(f"\n[Server] write_haiku called for topic '{topic}'")
+
+    async def work(task: ServerTaskContext) -> types.CallToolResult:
+        print(f"[Server] Task {task.task_id} starting sampling...")
+
+        result = await task.create_message(
+            messages=[
+                types.SamplingMessage(
+                    role="user",
+                    content=types.TextContent(type="text", text=f"Write a haiku about {topic}"),
+                )
+            ],
+            max_tokens=50,
+        )
 
-            result = await task.create_message(
-                messages=[
-                    types.SamplingMessage(
-                        role="user",
-                        content=types.TextContent(type="text", text=f"Write a haiku about {topic}"),
-                    )
-                ],
-                max_tokens=50,
-            )
+        haiku = "No response"
+        if isinstance(result.content, types.TextContent):
+            haiku = result.content.text
 
-            haiku = "No response"
-            if isinstance(result.content, types.TextContent):
-                haiku = result.content.text
+        print(f"[Server] Received sampling response: {haiku[:50]}...")
+        return types.CallToolResult(content=[types.TextContent(type="text", text=f"Haiku:\n{haiku}")])
 
-            print(f"[Server] Received sampling response: {haiku[:50]}...")
-            return types.CallToolResult(content=[types.TextContent(type="text", text=f"Haiku:\n{haiku}")])
+    return await ctx.experimental.run_task(work)
 
-        return await ctx.experimental.run_task(do_haiku)
 
+@server.call_tool()
+async def handle_call_tool(name: str, arguments: dict[str, Any]) -> types.CallToolResult | types.CreateTaskResult:
+    """Dispatch tool calls to their handlers."""
+    if name == "confirm_delete":
+        return await handle_confirm_delete(arguments)
+    elif name == "write_haiku":
+        return await handle_write_haiku(arguments)
     else:
         return types.CallToolResult(
             content=[types.TextContent(type="text", text=f"Unknown tool: {name}")],

examples/servers/simple-task/mcp_simple_task/server.py

@@ -32,8 +32,8 @@ async def list_tools() -> list[types.Tool]:
     ]
 
 
-@server.call_tool()
-async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[types.TextContent] | types.CreateTaskResult:
+async def handle_long_running_task(arguments: dict[str, Any]) -> types.CreateTaskResult:
+    """Handle the long_running_task tool - demonstrates status updates."""
     ctx = server.request_context
     ctx.experimental.validate_task_mode(types.TASK_REQUIRED)
 
@@ -52,6 +52,18 @@ async def work(task: ServerTaskContext) -> types.CallToolResult:
     return await ctx.experimental.run_task(work)
 
 
+@server.call_tool()
+async def handle_call_tool(name: str, arguments: dict[str, Any]) -> types.CallToolResult | types.CreateTaskResult:
+    """Dispatch tool calls to their handlers."""
+    if name == "long_running_task":
+        return await handle_long_running_task(arguments)
+    else:
+        return types.CallToolResult(
+            content=[types.TextContent(type="text", text=f"Unknown tool: {name}")],
+            isError=True,
+        )
+
+
 @click.command()
 @click.option("--port", default=8000, help="Port to listen on")
 def main(port: int) -> int: