feat: Add configurable timeout for tool execution

README.md

@@ -364,6 +364,59 @@ async def long_running_task(task_name: str, ctx: Context[ServerSession, None], s
 _Full example: [examples/snippets/servers/tool_progress.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/tool_progress.py)_
 <!-- /snippet-source -->
 
+#### Tool Timeouts
+
+FastMCP provides configurable timeouts for tool execution to prevent long-running tools from blocking indefinitely. By default, tools have a 300-second (5 minute) timeout, but this can be customized:
+
+```python
+from mcp.server.fastmcp import FastMCP
+
+# Set a custom timeout for all tools (in seconds)
+mcp = FastMCP("My Server", tool_timeout_seconds=60.0)  # 1 minute timeout
+
+# Disable timeout entirely (use with caution)
+mcp = FastMCP("My Server", tool_timeout_seconds=None)
+
+# Use the default 300 second timeout
+mcp = FastMCP("My Server")  # 5 minute default
+```
+
+When a tool exceeds its timeout, an `McpError` with error code `REQUEST_TIMEOUT` (-32001) is raised:
+
+```python
+from mcp.shared.exceptions import McpError
+from mcp.types import REQUEST_TIMEOUT
+
+@mcp.tool()
+async def slow_operation(data: str) -> str:
+    """A potentially slow operation."""
+    # If this takes longer than tool_timeout_seconds, it will be cancelled
+    result = await process_large_dataset(data)
+    return result
+
+# Clients can catch timeout errors
+async def call_tool_with_timeout():
+    try:
+        result = await session.call_tool("slow_operation", {"data": "..."})
+    except McpError as e:
+        if e.error.code == REQUEST_TIMEOUT:
+            print("Tool execution timed out")
+```
+
+**Configuration via environment variables:**
+
+```bash
+# Set timeout via environment variable
+FASTMCP_TOOL_TIMEOUT_SECONDS=120 python server.py
+```
+
+**Best practices:**
+
+- Choose timeouts based on expected tool execution time
+- Consider client-side timeouts as well for end-to-end timeout control
+- Log or monitor timeout occurrences to identify problematic tools
+- For truly long-running operations, consider using progress updates or async patterns
+
 #### Structured Output
 
 Tools will return structured results by default, if their return type
@@ -1072,6 +1125,7 @@ The FastMCP server instance accessible via `ctx.fastmcp` provides access to serv
   - `host` and `port` - Server network configuration
   - `mount_path`, `sse_path`, `streamable_http_path` - Transport paths
   - `stateless_http` - Whether the server operates in stateless mode
+  - `tool_timeout_seconds` - Maximum execution time for tools (default: 300 seconds)
   - And other configuration options
 
 ```python

src/mcp/server/fastmcp/server.py

@@ -112,6 +112,8 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
 
     # tool settings
     warn_on_duplicate_tools: bool
+    tool_timeout_seconds: float | None
+    """Maximum time in seconds for tool execution. None means no timeout. Default is 300 seconds (5 minutes)."""
 
     # prompt settings
     warn_on_duplicate_prompts: bool
@@ -168,6 +170,7 @@ def __init__(  # noqa: PLR0913
         warn_on_duplicate_resources: bool = True,
         warn_on_duplicate_tools: bool = True,
         warn_on_duplicate_prompts: bool = True,
+        tool_timeout_seconds: float | None = 300.0,
         dependencies: Collection[str] = (),
         lifespan: (Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None) = None,
         auth: AuthSettings | None = None,
@@ -187,6 +190,7 @@ def __init__(  # noqa: PLR0913
             warn_on_duplicate_resources=warn_on_duplicate_resources,
             warn_on_duplicate_tools=warn_on_duplicate_tools,
             warn_on_duplicate_prompts=warn_on_duplicate_prompts,
+            tool_timeout_seconds=tool_timeout_seconds,
             dependencies=list(dependencies),
             lifespan=lifespan,
             auth=auth,
@@ -202,7 +206,11 @@ def __init__(  # noqa: PLR0913
             # We need to create a Lifespan type that is a generic on the server type, like Starlette does.
             lifespan=(lifespan_wrapper(self, self.settings.lifespan) if self.settings.lifespan else default_lifespan),  # type: ignore
         )
-        self._tool_manager = ToolManager(tools=tools, warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools)
+        self._tool_manager = ToolManager(
+            tools=tools,
+            warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools,
+            timeout_seconds=self.settings.tool_timeout_seconds,
+        )
         self._resource_manager = ResourceManager(warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources)
         self._prompt_manager = PromptManager(warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts)
         # Validate auth configuration

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

@@ -6,13 +6,15 @@
 from functools import cached_property
 from typing import TYPE_CHECKING, Any
 
+import anyio
 from pydantic import BaseModel, Field
 
 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.shared.exceptions import McpError
 from mcp.shared.tool_name_validation import validate_and_warn_tool_name
-from mcp.types import Icon, ToolAnnotations
+from mcp.types import REQUEST_TIMEOUT, ErrorData, Icon, ToolAnnotations
 
 if TYPE_CHECKING:
     from mcp.server.fastmcp.server import Context
@@ -94,20 +96,45 @@ async def run(
         arguments: dict[str, Any],
         context: Context[ServerSessionT, LifespanContextT, RequestT] | None = None,
         convert_result: bool = False,
+        timeout_seconds: float | None = None,
     ) -> Any:
-        """Run the tool with arguments."""
-        try:
-            result = await self.fn_metadata.call_fn_with_arg_validation(
-                self.fn,
-                self.is_async,
-                arguments,
-                {self.context_kwarg: context} if self.context_kwarg is not None else None,
-            )
+        """Run the tool with arguments.
+
+        Args:
+            arguments: The arguments to pass to the tool function
+            context: Optional context to inject into the tool function
+            convert_result: Whether to convert the result to MCP types
+            timeout_seconds: Maximum execution time in seconds. None means no timeout.
 
-            if convert_result:
-                result = self.fn_metadata.convert_result(result)
+        Returns:
+            The result of the tool execution
 
-            return result
+        Raises:
+            McpError: If the tool execution times out (REQUEST_TIMEOUT error code)
+            ToolError: If the tool execution fails for other reasons
+        """
+        try:
+            # Wrap execution in timeout if configured
+            with anyio.fail_after(timeout_seconds):
+                result = await self.fn_metadata.call_fn_with_arg_validation(
+                    self.fn,
+                    self.is_async,
+                    arguments,
+                    {self.context_kwarg: context} if self.context_kwarg is not None else None,
+                )
+
+                if convert_result:
+                    result = self.fn_metadata.convert_result(result)
+
+                return result
+        except TimeoutError as e:
+            # Convert timeout to MCP error with REQUEST_TIMEOUT code
+            raise McpError(
+                ErrorData(
+                    code=REQUEST_TIMEOUT,
+                    message=f"Tool '{self.name}' execution exceeded timeout of {timeout_seconds} seconds",
+                )
+            ) from e
         except Exception as e:
             raise ToolError(f"Error executing tool {self.name}: {e}") from e
 

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

@@ -22,6 +22,7 @@ class ToolManager:
     def __init__(
         self,
         warn_on_duplicate_tools: bool = True,
+        timeout_seconds: float | None = None,
         *,
         tools: list[Tool] | None = None,
     ):
@@ -33,6 +34,7 @@ def __init__(
                 self._tools[tool.name] = tool
 
         self.warn_on_duplicate_tools = warn_on_duplicate_tools
+        self.timeout_seconds = timeout_seconds
 
     def get_tool(self, name: str) -> Tool | None:
         """Get tool by name."""
@@ -90,4 +92,9 @@ async def call_tool(
         if not tool:
             raise ToolError(f"Unknown tool: {name}")
 
-        return await tool.run(arguments, context=context, convert_result=convert_result)
+        return await tool.run(
+            arguments,
+            context=context,
+            convert_result=convert_result,
+            timeout_seconds=self.timeout_seconds,
+        )

src/mcp/types.py

@@ -152,7 +152,8 @@ class JSONRPCResponse(BaseModel):
 
 # SDK error codes
 CONNECTION_CLOSED = -32000
-# REQUEST_TIMEOUT = -32001  # the typescript sdk uses this
+REQUEST_TIMEOUT = -32001
+"""Error code indicating that a request exceeded the configured timeout period."""
 
 # Standard JSON-RPC error codes
 PARSE_ERROR = -32700

tests/server/fastmcp/test_tool_manager.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 import logging
 from dataclasses import dataclass
@@ -12,7 +13,8 @@
 from mcp.server.fastmcp.utilities.func_metadata import ArgModelBase, FuncMetadata
 from mcp.server.session import ServerSessionT
 from mcp.shared.context import LifespanContextT, RequestT
-from mcp.types import TextContent, ToolAnnotations
+from mcp.shared.exceptions import McpError
+from mcp.types import REQUEST_TIMEOUT, TextContent, ToolAnnotations
 
 
 class TestAddTools:
@@ -920,3 +922,163 @@ def test_func() -> str:  # pragma: no cover
         # Remove with correct case
         manager.remove_tool("test_func")
         assert manager.get_tool("test_func") is None
+
+
+class TestToolTimeout:
+    """Test timeout behavior for tool execution."""
+
+    @pytest.mark.anyio
+    async def test_tool_timeout_exceeded(self):
+        """Test that a slow tool times out and raises McpError with REQUEST_TIMEOUT code."""
+
+        async def slow_tool(duration: float) -> str:  # pragma: no cover
+            """A tool that sleeps for the specified duration."""
+            await asyncio.sleep(duration)
+            return "completed"
+
+        manager = ToolManager(timeout_seconds=0.1)  # 100ms timeout
+        manager.add_tool(slow_tool)
+
+        # Tool should timeout after 100ms
+        with pytest.raises(McpError) as exc_info:
+            await manager.call_tool("slow_tool", {"duration": 1.0})  # Try to sleep for 1 second
+
+        # Verify the error code is REQUEST_TIMEOUT
+        assert exc_info.value.error.code == REQUEST_TIMEOUT
+        assert "slow_tool" in exc_info.value.error.message
+        assert "exceeded timeout" in exc_info.value.error.message
+
+    @pytest.mark.anyio
+    async def test_tool_completes_before_timeout(self):
+        """Test that a fast tool completes successfully before timeout."""
+
+        async def fast_tool(value: str) -> str:
+            """A tool that completes quickly."""
+            await asyncio.sleep(0.01)  # 10ms
+            return f"processed: {value}"
+
+        manager = ToolManager(timeout_seconds=1.0)  # 1 second timeout
+        manager.add_tool(fast_tool)
+
+        # Tool should complete successfully
+        result = await manager.call_tool("fast_tool", {"value": "test"})
+        assert result == "processed: test"
+
+    @pytest.mark.anyio
+    async def test_tool_without_timeout(self):
+        """Test that tools work normally when timeout is None."""
+
+        async def slow_tool(duration: float) -> str:
+            """A tool that can take any amount of time."""
+            await asyncio.sleep(duration)
+            return "completed"
+
+        manager = ToolManager(timeout_seconds=None)  # No timeout
+        manager.add_tool(slow_tool)
+
+        # Tool should complete without timeout even if slow
+        result = await manager.call_tool("slow_tool", {"duration": 0.2})
+        assert result == "completed"
+
+    @pytest.mark.anyio
+    @pytest.mark.skip(
+        reason="Blocking sync operations (time.sleep) don't respect anyio.fail_after() timeouts. "
+        "Use anyio.sleep() in async functions for timeout support."
+    )
+    async def test_sync_tool_timeout(self):  # pragma: no cover
+        """Test that synchronous tools also respect timeout."""
+        import time
+
+        def slow_sync_tool(duration: float) -> str:
+            """A synchronous tool that sleeps."""
+            time.sleep(duration)
+            return "completed"
+
+        manager = ToolManager(timeout_seconds=0.1)  # 100ms timeout
+        manager.add_tool(slow_sync_tool)
+
+        # Sync tool should also timeout
+        with pytest.raises(McpError) as exc_info:
+            await manager.call_tool("slow_sync_tool", {"duration": 1.0})
+
+        assert exc_info.value.error.code == REQUEST_TIMEOUT
+
+    @pytest.mark.anyio
+    async def test_timeout_with_context_injection(self):
+        """Test that timeout works correctly with context injection."""
+
+        async def slow_tool_with_context(
+            duration: float, ctx: Context[ServerSessionT, None]
+        ) -> str:  # pragma: no cover
+            """A tool with context that times out."""
+            await asyncio.sleep(duration)
+            return "completed"
+
+        manager = ToolManager(timeout_seconds=0.1)
+        manager.add_tool(slow_tool_with_context)
+
+        mcp = FastMCP()
+        ctx = mcp.get_context()
+
+        # Tool should timeout even with context injection
+        with pytest.raises(McpError) as exc_info:
+            await manager.call_tool("slow_tool_with_context", {"duration": 1.0}, context=ctx)
+
+        assert exc_info.value.error.code == REQUEST_TIMEOUT
+
+    @pytest.mark.anyio
+    async def test_tool_error_not_confused_with_timeout(self):
+        """Test that regular tool errors are not confused with timeout errors."""
+
+        async def failing_tool(should_fail: bool) -> str:
+            """A tool that raises an error."""
+            if should_fail:
+                raise ValueError("Tool failed intentionally")
+            return "success"  # pragma: no cover
+
+        manager = ToolManager(timeout_seconds=1.0)
+        manager.add_tool(failing_tool)
+
+        # Regular errors should still be ToolError, not timeout
+        with pytest.raises(ToolError, match="Error executing tool failing_tool"):
+            await manager.call_tool("failing_tool", {"should_fail": True})
+
+    @pytest.mark.anyio
+    async def test_fastmcp_timeout_setting(self):
+        """Test that FastMCP passes timeout setting to ToolManager."""
+
+        async def slow_tool() -> str:  # pragma: no cover
+            """A slow tool."""
+            await asyncio.sleep(1.0)
+            return "completed"
+
+        # Create FastMCP with custom timeout
+        app = FastMCP(tool_timeout_seconds=0.1)
+
+        @app.tool()
+        async def test_tool() -> str:  # pragma: no cover
+            """Test tool."""
+            await asyncio.sleep(1.0)
+            return "completed"
+
+        # Tool should timeout based on FastMCP setting
+        with pytest.raises(McpError) as exc_info:
+            await app._tool_manager.call_tool("test_tool", {})
+
+        assert exc_info.value.error.code == REQUEST_TIMEOUT
+
+    @pytest.mark.anyio
+    async def test_fastmcp_no_timeout(self):
+        """Test that FastMCP works with timeout disabled."""
+
+        app = FastMCP(tool_timeout_seconds=None)
+
+        @app.tool()
+        async def slow_tool() -> str:
+            """A slow tool."""
+            await asyncio.sleep(0.2)
+            return "completed"
+
+        # Tool should complete without timeout
+        result = await app._tool_manager.call_tool("slow_tool", {})
+        assert result == "completed"