diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml
index dd6af5154..ec4457e24 100644
--- a/.github/workflows/run-tests.yml
+++ b/.github/workflows/run-tests.yml
@@ -46,6 +46,7 @@ jobs:
           enable-cache: true
           cache-dependency-glob: "uv.lock"
           python-version: ${{ matrix.python-version }}
+
       - name: Install FastMCP
         run: uv sync --dev --locked
 
diff --git a/docs/clients/client.mdx b/docs/clients/client.mdx
index 1f1e6aa76..4b77c5008 100644
--- a/docs/clients/client.mdx
+++ b/docs/clients/client.mdx
@@ -115,14 +115,18 @@ The standard client methods return user-friendly representations that may change
     tools = await client.list_tools()
     # tools -> list[mcp.types.Tool]
     ```
-*   **`call_tool(name: str, arguments: dict[str, Any] | None = None)`**: Executes a tool on the server.
+*   **`call_tool(name: str, arguments: dict[str, Any] | None = None, timeout: float | None = None)`**: Executes a tool on the server.
     ```python
     result = await client.call_tool("add", {"a": 5, "b": 3})
     # result -> list[mcp.types.TextContent | mcp.types.ImageContent | ...]
     print(result[0].text) # Assuming TextContent, e.g., '8'
+    
+    # With timeout (aborts if execution takes longer than 2 seconds)
+    result = await client.call_tool("long_running_task", {"param": "value"}, timeout=2.0)
     ```
     *   Arguments are passed as a dictionary. FastMCP servers automatically handle JSON string parsing for complex types if needed.
     *   Returns a list of content objects (usually `TextContent` or `ImageContent`).
+    *   The optional `timeout` parameter limits the maximum execution time (in seconds) for this specific call, overriding any client-level timeout.
 
 #### Resource Operations
 
@@ -191,6 +195,45 @@ These methods are especially useful for debugging or when you need to access met
 
 MCP allows servers to interact with clients in order to provide additional capabilities. The `Client` constructor accepts additional configuration to handle these server requests.
 
+#### Timeout Control
+
+<VersionBadge version="2.3.4" />
+
+You can control request timeouts at both the client level and individual request level:
+
+```python
+from fastmcp import Client
+from fastmcp.exceptions import McpError
+
+# Client with a global 5-second timeout for all requests
+client = Client(
+    my_mcp_server,
+    timeout=5.0  # Default timeout in seconds
+)
+
+async with client:
+    # This uses the global 5-second timeout
+    result1 = await client.call_tool("quick_task", {"param": "value"})
+    
+    # This specifies a 10-second timeout for this specific call
+    result2 = await client.call_tool("slow_task", {"param": "value"}, timeout=10.0)
+    
+    try:
+        # This will likely timeout
+        result3 = await client.call_tool("medium_task", {"param": "value"}, timeout=0.01)
+    except McpError as e:
+        # Handle timeout error
+        print(f"The task timed out: {e}")
+```
+
+<Warning>
+Timeout behavior varies between transport types:
+
+- With **SSE** transport, the per-request (tool call) timeout **always** takes precedence, regardless of which is lower.
+- With **HTTP** transport, the **lower** of the two timeouts (client or tool call) takes precedence.
+
+For consistent behavior across all transports, we recommend explicitly setting timeouts at the individual tool call level when needed, rather than relying on client-level timeouts.
+</Warning>
 
 #### LLM Sampling
 
diff --git a/src/fastmcp/client/client.py b/src/fastmcp/client/client.py
index 6c56893a0..9f5a111f9 100644
--- a/src/fastmcp/client/client.py
+++ b/src/fastmcp/client/client.py
@@ -35,8 +35,35 @@ class Client:
     """
     MCP client that delegates connection management to a Transport instance.
 
-    The Client class is primarily concerned with MCP protocol logic,
-    while the Transport handles connection establishment and management.
+    The Client class is responsible for MCP protocol logic, while the Transport
+    handles connection establishment and management. Client provides methods
+    for working with resources, prompts, tools and other MCP capabilities.
+
+    Args:
+        transport: Connection source specification, which can be:
+            - ClientTransport: Direct transport instance
+            - FastMCP: In-process FastMCP server
+            - AnyUrl | str: URL to connect to
+            - Path: File path for local socket
+            - dict: Transport configuration
+        roots: Optional RootsList or RootsHandler for filesystem access
+        sampling_handler: Optional handler for sampling requests
+        log_handler: Optional handler for log messages
+        message_handler: Optional handler for protocol messages
+        timeout: Optional timeout for requests (seconds or timedelta)
+
+    Examples:
+        ```python
+        # Connect to FastMCP server
+        client = Client("http://localhost:8080")
+
+        async with client:
+            # List available resources
+            resources = await client.list_resources()
+
+            # Call a tool
+            result = await client.call_tool("my_tool", {"param": "value"})
+        ```
     """
 
     def __init__(
@@ -47,19 +74,22 @@ class Client:
         sampling_handler: SamplingHandler | None = None,
         log_handler: LogHandler | None = None,
         message_handler: MessageHandler | None = None,
-        read_timeout_seconds: datetime.timedelta | None = None,
+        timeout: datetime.timedelta | float | int | None = None,
     ):
         self.transport = infer_transport(transport)
         self._session: ClientSession | None = None
         self._exit_stack: AsyncExitStack | None = None
         self._nesting_counter: int = 0
 
+        if isinstance(timeout, int | float):
+            timeout = datetime.timedelta(seconds=timeout)
+
         self._session_kwargs: SessionKwargs = {
             "sampling_callback": None,
             "list_roots_callback": None,
             "logging_callback": log_handler,
             "message_handler": message_handler,
-            "read_timeout_seconds": read_timeout_seconds,
+            "read_timeout_seconds": timeout,
         }
 
         if roots is not None:
@@ -397,7 +427,10 @@ class Client:
     # --- Call Tool ---
 
     async def call_tool_mcp(
-        self, name: str, arguments: dict[str, Any]
+        self,
+        name: str,
+        arguments: dict[str, Any],
+        timeout: datetime.timedelta | float | int | None = None,
     ) -> mcp.types.CallToolResult:
         """Send a tools/call request and return the complete MCP protocol result.
 
@@ -407,7 +440,7 @@ class Client:
         Args:
             name (str): The name of the tool to call.
             arguments (dict[str, Any]): Arguments to pass to the tool.
-
+            timeout (datetime.timedelta | float | int | None, optional): The timeout for the tool call. Defaults to None.
         Returns:
             mcp.types.CallToolResult: The complete response object from the protocol,
                 containing the tool result and any additional metadata.
@@ -415,13 +448,19 @@ class Client:
         Raises:
             RuntimeError: If called while the client is not connected.
         """
-        result = await self.session.call_tool(name=name, arguments=arguments)
+
+        if isinstance(timeout, int | float):
+            timeout = datetime.timedelta(seconds=timeout)
+        result = await self.session.call_tool(
+            name=name, arguments=arguments, read_timeout_seconds=timeout
+        )
         return result
 
     async def call_tool(
         self,
         name: str,
         arguments: dict[str, Any] | None = None,
+        timeout: datetime.timedelta | float | int | None = None,
     ) -> list[
         mcp.types.TextContent | mcp.types.ImageContent | mcp.types.EmbeddedResource
     ]:
@@ -441,7 +480,11 @@ class Client:
             ToolError: If the tool call results in an error.
             RuntimeError: If called while the client is not connected.
         """
-        result = await self.call_tool_mcp(name=name, arguments=arguments or {})
+        result = await self.call_tool_mcp(
+            name=name,
+            arguments=arguments or {},
+            timeout=timeout,
+        )
         if result.isError:
             msg = cast(mcp.types.TextContent, result.content[0]).text
             raise ToolError(msg)
diff --git a/src/fastmcp/client/transports.py b/src/fastmcp/client/transports.py
index f201a1c6b..1b559a7c4 100644
--- a/src/fastmcp/client/transports.py
+++ b/src/fastmcp/client/transports.py
@@ -8,7 +8,7 @@ import sys
 import warnings
 from collections.abc import AsyncIterator
 from pathlib import Path
-from typing import Any, TypedDict
+from typing import Any, TypedDict, cast
 
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.session import (
@@ -102,7 +102,12 @@ class WSTransport(ClientTransport):
 class SSETransport(ClientTransport):
     """Transport implementation that connects to an MCP server via Server-Sent Events."""
 
-    def __init__(self, url: str | AnyUrl, headers: dict[str, str] | None = None):
+    def __init__(
+        self,
+        url: str | AnyUrl,
+        headers: dict[str, str] | None = None,
+        sse_read_timeout: datetime.timedelta | float | int | None = None,
+    ):
         if isinstance(url, AnyUrl):
             url = str(url)
         if not isinstance(url, str) or not url.startswith("http"):
@@ -110,11 +115,28 @@ class SSETransport(ClientTransport):
         self.url = url
         self.headers = headers or {}
 
+        if isinstance(sse_read_timeout, int | float):
+            sse_read_timeout = datetime.timedelta(seconds=sse_read_timeout)
+        self.sse_read_timeout = sse_read_timeout
+
     @contextlib.asynccontextmanager
     async def connect_session(
         self, **session_kwargs: Unpack[SessionKwargs]
     ) -> AsyncIterator[ClientSession]:
-        async with sse_client(self.url, headers=self.headers) as transport:
+        client_kwargs = {}
+        # sse_read_timeout has a default value set, so we can't pass None without overriding it
+        # instead we simply leave the kwarg out if it's not provided
+        if self.sse_read_timeout is not None:
+            client_kwargs["sse_read_timeout"] = self.sse_read_timeout.total_seconds()
+        if session_kwargs.get("read_timeout_seconds", None) is not None:
+            read_timeout_seconds = cast(
+                datetime.timedelta, session_kwargs.get("read_timeout_seconds")
+            )
+            client_kwargs["timeout"] = read_timeout_seconds.total_seconds()
+
+        async with sse_client(
+            self.url, headers=self.headers, **client_kwargs
+        ) as transport:
             read_stream, write_stream = transport
             async with ClientSession(
                 read_stream, write_stream, **session_kwargs
@@ -129,7 +151,12 @@ class SSETransport(ClientTransport):
 class StreamableHttpTransport(ClientTransport):
     """Transport implementation that connects to an MCP server via Streamable HTTP Requests."""
 
-    def __init__(self, url: str | AnyUrl, headers: dict[str, str] | None = None):
+    def __init__(
+        self,
+        url: str | AnyUrl,
+        headers: dict[str, str] | None = None,
+        sse_read_timeout: datetime.timedelta | float | int | None = None,
+    ):
         if isinstance(url, AnyUrl):
             url = str(url)
         if not isinstance(url, str) or not url.startswith("http"):
@@ -137,11 +164,25 @@ class StreamableHttpTransport(ClientTransport):
         self.url = url
         self.headers = headers or {}
 
+        if isinstance(sse_read_timeout, int | float):
+            sse_read_timeout = datetime.timedelta(seconds=sse_read_timeout)
+        self.sse_read_timeout = sse_read_timeout
+
     @contextlib.asynccontextmanager
     async def connect_session(
         self, **session_kwargs: Unpack[SessionKwargs]
     ) -> AsyncIterator[ClientSession]:
-        async with streamablehttp_client(self.url, headers=self.headers) as transport:
+        client_kwargs = {}
+        # sse_read_timeout has a default value set, so we can't pass None without overriding it
+        # instead we simply leave the kwarg out if it's not provided
+        if self.sse_read_timeout is not None:
+            client_kwargs["sse_read_timeout"] = self.sse_read_timeout
+        if session_kwargs.get("read_timeout_seconds", None) is not None:
+            client_kwargs["timeout"] = session_kwargs.get("read_timeout_seconds")
+
+        async with streamablehttp_client(
+            self.url, headers=self.headers, **client_kwargs
+        ) as transport:
             read_stream, write_stream, _ = transport
             async with ClientSession(
                 read_stream, write_stream, **session_kwargs
diff --git a/src/fastmcp/exceptions.py b/src/fastmcp/exceptions.py
index c105cce8e..24864d8a0 100644
--- a/src/fastmcp/exceptions.py
+++ b/src/fastmcp/exceptions.py
@@ -1,5 +1,7 @@
 """Custom exceptions for FastMCP."""
 
+from mcp import McpError  # noqa: F401
+
 
 class FastMCPError(Exception):
     """Base error for FastMCP."""
diff --git a/src/fastmcp/utilities/exceptions.py b/src/fastmcp/utilities/exceptions.py
index eaba30ee6..e50dc57b0 100644
--- a/src/fastmcp/utilities/exceptions.py
+++ b/src/fastmcp/utilities/exceptions.py
@@ -1,7 +1,10 @@
 from collections.abc import Callable, Iterable, Mapping
 from typing import Any
 
+import httpx
+import mcp.types
 from exceptiongroup import BaseExceptionGroup
+from mcp import McpError
 
 import fastmcp
 
@@ -16,12 +19,19 @@ def iter_exc(group: BaseExceptionGroup):
 
 def _exception_handler(group: BaseExceptionGroup):
     for leaf in iter_exc(group):
+        if isinstance(leaf, httpx.ConnectTimeout):
+            raise McpError(
+                error=mcp.types.ErrorData(
+                    code=httpx.codes.REQUEST_TIMEOUT,
+                    message="Timed out while waiting for response.",
+                )
+            )
         raise leaf
 
 
 # this catch handler is used to catch taskgroup exception groups and raise the
 # first exception. This allows more sane debugging.
-catch_handlers: Mapping[
+_catch_handlers: Mapping[
     type[BaseException] | Iterable[type[BaseException]],
     Callable[[BaseExceptionGroup[Any]], Any],
 ] = {
@@ -34,6 +44,6 @@ def get_catch_handlers() -> Mapping[
     Callable[[BaseExceptionGroup[Any]], Any],
 ]:
     if fastmcp.settings.settings.client_raise_first_exceptiongroup_error:
-        return catch_handlers
+        return _catch_handlers
     else:
         return {}
diff --git a/tests/client/test_client.py b/tests/client/test_client.py
index ff460b295..1d8c3187d 100644
--- a/tests/client/test_client.py
+++ b/tests/client/test_client.py
@@ -1,6 +1,8 @@
+import asyncio
 from typing import cast
 
 import pytest
+from mcp import McpError
 from pydantic import AnyUrl
 
 from fastmcp.client import Client
@@ -27,6 +29,12 @@ def fastmcp_server():
         """Add two numbers together."""
         return a + b
 
+    @server.tool()
+    async def sleep(seconds: float) -> str:
+        """Sleep for a given number of seconds."""
+        await asyncio.sleep(seconds)
+        return f"Slept for {seconds} seconds"
+
     # Add a resource
     @server.resource(uri="data://users")
     async def get_users():
@@ -78,8 +86,8 @@ async def test_list_tools(fastmcp_server):
         result = await client.list_tools()
 
         # Check that our tools are available
-        assert len(result) == 2
-        assert set(tool.name for tool in result) == {"greet", "add"}
+        assert len(result) == 3
+        assert set(tool.name for tool in result) == {"greet", "add", "sleep"}
 
 
 async def test_list_tools_mcp(fastmcp_server):
@@ -91,8 +99,8 @@ async def test_list_tools_mcp(fastmcp_server):
 
         # Check that we got the raw MCP ListToolsResult object
         assert hasattr(result, "tools")
-        assert len(result.tools) == 2
-        assert set(tool.name for tool in result.tools) == {"greet", "add"}
+        assert len(result.tools) == 3
+        assert set(tool.name for tool in result.tools) == {"greet", "add", "sleep"}
 
 
 async def test_call_tool(fastmcp_server):
@@ -499,3 +507,39 @@ class TestErrorHandling:
             with pytest.raises(Exception) as excinfo:
                 await client.read_resource(AnyUrl("error://resource/123"))
             assert "This is a resource error (xyz)" in str(excinfo.value)
+
+
+class TestTimeout:
+    async def test_timeout(self, fastmcp_server: FastMCP):
+        async with Client(
+            transport=FastMCPTransport(fastmcp_server), timeout=0.01
+        ) as client:
+            with pytest.raises(
+                McpError,
+                match="Timed out while waiting for response to ClientRequest. Waited 0.01 seconds",
+            ):
+                await client.call_tool("sleep", {"seconds": 0.1})
+
+    async def test_timeout_tool_call(self, fastmcp_server: FastMCP):
+        async with Client(transport=FastMCPTransport(fastmcp_server)) as client:
+            with pytest.raises(McpError):
+                await client.call_tool("sleep", {"seconds": 0.1}, timeout=0.01)
+
+    async def test_timeout_tool_call_overrides_client_timeout(
+        self, fastmcp_server: FastMCP
+    ):
+        async with Client(
+            transport=FastMCPTransport(fastmcp_server),
+            timeout=2,
+        ) as client:
+            with pytest.raises(McpError):
+                await client.call_tool("sleep", {"seconds": 0.1}, timeout=0.01)
+
+    async def test_timeout_tool_call_overrides_client_timeout_even_if_lower(
+        self, fastmcp_server: FastMCP
+    ):
+        async with Client(
+            transport=FastMCPTransport(fastmcp_server),
+            timeout=0.01,
+        ) as client:
+            await client.call_tool("sleep", {"seconds": 0.1}, timeout=2)
diff --git a/tests/client/test_sse.py b/tests/client/test_sse.py
index 3259fb7d5..f70395550 100644
--- a/tests/client/test_sse.py
+++ b/tests/client/test_sse.py
@@ -1,9 +1,11 @@
+import asyncio
 import json
 import sys
 from collections.abc import Generator
 
 import pytest
 import uvicorn
+from mcp import McpError
 from mcp.types import TextResourceContents
 from starlette.applications import Starlette
 from starlette.routing import Mount
@@ -31,6 +33,12 @@ def fastmcp_server():
         """Add two numbers together."""
         return a + b
 
+    @server.tool()
+    async def sleep(seconds: float) -> str:
+        """Sleep for a given number of seconds."""
+        await asyncio.sleep(seconds)
+        return f"Slept for {seconds} seconds"
+
     # Add a resource
     @server.resource(uri="data://users")
     async def get_users():
@@ -126,3 +134,49 @@ async def test_nested_sse_server_resolves_correctly():
         ) as client:
             result = await client.ping()
             assert result is True
+
+
+class TestTimeout:
+    async def test_timeout(self, sse_server: str):
+        with pytest.raises(
+            McpError,
+            match="Timed out while waiting for response to ClientRequest. Waited 0.01 seconds",
+        ):
+            async with Client(
+                transport=SSETransport(sse_server),
+                timeout=0.01,
+            ) as client:
+                await client.call_tool("sleep", {"seconds": 0.1})
+
+    async def test_timeout_tool_call(self, sse_server: str):
+        async with Client(transport=SSETransport(sse_server)) as client:
+            with pytest.raises(McpError, match="Timed out"):
+                await client.call_tool("sleep", {"seconds": 0.1}, timeout=0.01)
+
+    async def test_timeout_tool_call_overrides_client_timeout_if_lower(
+        self, sse_server: str
+    ):
+        async with Client(
+            transport=SSETransport(sse_server),
+            timeout=2,
+        ) as client:
+            with pytest.raises(McpError, match="Timed out"):
+                await client.call_tool("sleep", {"seconds": 0.1}, timeout=0.01)
+
+    @pytest.mark.skipif(
+        sys.platform == "win32",
+        reason="This test is flaky on Windows. Sometimes the client timeout is respected and sometimes it is not.",
+    )
+    async def test_timeout_client_timeout_does_not_override_tool_call_timeout_if_lower(
+        self, sse_server: str
+    ):
+        """
+        With SSE, the tool call timeout always takes precedence over the client.
+
+        Note: on Windows, the behavior appears unpredictable.
+        """
+        async with Client(
+            transport=SSETransport(sse_server),
+            timeout=0.01,
+        ) as client:
+            await client.call_tool("sleep", {"seconds": 0.1}, timeout=2)
diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py
index a388cc3cb..9b1528c5d 100644
--- a/tests/client/test_streamable_http.py
+++ b/tests/client/test_streamable_http.py
@@ -1,9 +1,11 @@
+import asyncio
 import json
 import sys
 from collections.abc import Generator
 
 import pytest
 import uvicorn
+from mcp import McpError
 from mcp.types import TextResourceContents
 from starlette.applications import Starlette
 from starlette.routing import Mount
@@ -31,6 +33,12 @@ def fastmcp_server():
         """Add two numbers together."""
         return a + b
 
+    @server.tool()
+    async def sleep(seconds: float) -> str:
+        """Sleep for a given number of seconds."""
+        await asyncio.sleep(seconds)
+        return f"Slept for {seconds} seconds"
+
     # Add a resource
     @server.resource(uri="data://users")
     async def get_users():
@@ -139,3 +147,42 @@ async def test_nested_streamable_http_server_resolves_correctly():
         ) as client:
             result = await client.ping()
             assert result is True
+
+
+class TestTimeout:
+    async def test_timeout(self, streamable_http_server: str):
+        # note this transport behaves differently than others and raises
+        # McpError from the *client* context
+        with pytest.raises(McpError, match="Timed out"):
+            async with Client(
+                transport=StreamableHttpTransport(streamable_http_server),
+                timeout=0.01,
+            ) as client:
+                await client.call_tool("sleep", {"seconds": 0.1})
+
+    async def test_timeout_tool_call(self, streamable_http_server: str):
+        async with Client(
+            transport=StreamableHttpTransport(streamable_http_server),
+        ) as client:
+            with pytest.raises(McpError):
+                await client.call_tool("sleep", {"seconds": 0.1}, timeout=0.01)
+
+    async def test_timeout_tool_call_overrides_client_timeout(
+        self, streamable_http_server: str
+    ):
+        async with Client(
+            transport=StreamableHttpTransport(streamable_http_server),
+            timeout=2,
+        ) as client:
+            with pytest.raises(McpError):
+                await client.call_tool("sleep", {"seconds": 0.1}, timeout=0.01)
+
+    async def test_timeout_client_timeout_overrides_tool_call_timeout_if_lower(
+        self, streamable_http_server: str
+    ):
+        with pytest.raises(McpError):
+            async with Client(
+                transport=StreamableHttpTransport(streamable_http_server),
+                timeout=0.01,
+            ) as client:
+                await client.call_tool("sleep", {"seconds": 0.1}, timeout=2)