diff --git a/src/fastmcp/server/providers/fastmcp_provider.py b/src/fastmcp/server/providers/fastmcp_provider.py
index ce0a7ad7b3..ee345bd9e5 100644
--- a/src/fastmcp/server/providers/fastmcp_provider.py
+++ b/src/fastmcp/server/providers/fastmcp_provider.py
@@ -22,6 +22,7 @@
 from fastmcp.resources.resource import Resource, ResourceResult
 from fastmcp.resources.template import ResourceTemplate
 from fastmcp.server.providers.base import Provider
+from fastmcp.server.tasks.config import TaskMeta
 from fastmcp.tools.tool import Tool, ToolResult
 from fastmcp.utilities.components import FastMCPComponent
 
@@ -85,24 +86,31 @@ def wrap(cls, server: Any, tool: Tool) -> FastMCPProviderTool:
         )
 
     async def _run(
-        self, arguments: dict[str, Any]
+        self,
+        arguments: dict[str, Any],
+        task_meta: TaskMeta | None = None,
     ) -> ToolResult | mcp.types.CreateTaskResult:
-        """Skip task handling - delegate to run() which calls child middleware.
+        """Delegate to child server's call_tool() with task_meta.
 
-        The actual underlying tool will check _task_metadata contextvar and
-        submit to Docket if appropriate. This wrapper just passes through.
+        Passes task_meta through to the child server so it can handle
+        backgrounding appropriately.
         """
-        return await self.run(arguments)
+        return await self._server.call_tool(
+            self._original_name, arguments, task_meta=task_meta
+        )
 
     async def run(
         self, arguments: dict[str, Any]
     ) -> ToolResult | mcp.types.CreateTaskResult:  # type: ignore[override]
-        """Delegate to child server's call_tool().
+        """Not implemented - use _run() which delegates to child server.
 
-        This runs BEFORE any backgrounding decision - the actual underlying
-        tool will check contextvars and submit to Docket if appropriate.
+        FastMCPProviderTool._run() handles all execution by delegating
+        to the child server's call_tool() with task_meta.
         """
-        return await self._server.call_tool(self._original_name, arguments)
+        raise NotImplementedError(
+            "FastMCPProviderTool.run() should not be called directly. "
+            "Use _run() which delegates to the child server's call_tool()."
+        )
 
 
 class FastMCPProviderResource(Resource):
diff --git a/src/fastmcp/server/server.py b/src/fastmcp/server/server.py
index 816f30dc3e..eb161f7510 100644
--- a/src/fastmcp/server/server.py
+++ b/src/fastmcp/server/server.py
@@ -79,7 +79,7 @@
 from fastmcp.server.middleware import Middleware, MiddlewareContext
 from fastmcp.server.providers import LocalProvider, Provider
 from fastmcp.server.tasks.capabilities import get_task_capabilities
-from fastmcp.server.tasks.config import TaskConfig
+from fastmcp.server.tasks.config import TaskConfig, TaskMeta
 from fastmcp.settings import DuplicateBehavior as DuplicateBehaviorSetting
 from fastmcp.settings import Settings
 from fastmcp.tools.tool import FunctionTool, Tool, ToolResult
@@ -1126,12 +1126,33 @@ async def get_component(
 
         raise NotFoundError(f"Unknown component: {key}")
 
+    @overload
+    async def call_tool(
+        self,
+        name: str,
+        arguments: dict[str, Any] | None = None,
+        *,
+        run_middleware: bool = True,
+        task_meta: None = None,
+    ) -> ToolResult: ...
+
+    @overload
+    async def call_tool(
+        self,
+        name: str,
+        arguments: dict[str, Any] | None = None,
+        *,
+        run_middleware: bool = True,
+        task_meta: TaskMeta,
+    ) -> mcp.types.CreateTaskResult: ...
+
     async def call_tool(
         self,
         name: str,
         arguments: dict[str, Any] | None = None,
         *,
         run_middleware: bool = True,
+        task_meta: TaskMeta | None = None,
     ) -> ToolResult | mcp.types.CreateTaskResult:
         """Call a tool by name.
 
@@ -1142,16 +1163,23 @@ async def call_tool(
             arguments: Tool arguments (optional)
             run_middleware: If True (default), apply the middleware chain.
                 Set to False when called from middleware to avoid re-applying.
+            task_meta: If provided, execute as a background task and return
+                CreateTaskResult. If None (default), execute synchronously and
+                return ToolResult.
 
         Returns:
-            ToolResult with content and optional structured_content.
-            May return CreateTaskResult if called in MCP context with task metadata.
+            ToolResult when task_meta is None.
+            CreateTaskResult when task_meta is provided.
 
         Raises:
             NotFoundError: If tool not found or disabled
             ToolError: If tool execution fails
             ValidationError: If arguments fail validation
         """
+        # Enrich task_meta with fn_key if task execution requested
+        if task_meta is not None and task_meta.fn_key is None:
+            task_meta = TaskMeta(ttl=task_meta.ttl, fn_key=Tool.make_key(name))
+
         async with fastmcp.server.context.Context(fastmcp=self) as ctx:
             if run_middleware:
                 mw_context = MiddlewareContext[CallToolRequestParams](
@@ -1169,6 +1197,7 @@ async def call_tool(
                         context.message.name,
                         context.message.arguments or {},
                         run_middleware=False,
+                        task_meta=task_meta,
                     ),
                 )
 
@@ -1177,7 +1206,7 @@ async def call_tool(
                 tool = await provider.get_tool(name)
                 if tool is not None and self._is_component_enabled(tool):
                     try:
-                        return await tool._run(arguments or {})
+                        return await tool._run(arguments or {}, task_meta=task_meta)
                     except FastMCPError:
                         logger.exception(f"Error calling tool {name!r}")
                         raise
@@ -1491,8 +1520,9 @@ async def _call_tool_mcp(
         """
         Handle MCP 'callTool' requests.
 
-        Sets task metadata contextvar and calls call_tool(). The tool's _run() method
-        handles the backgrounding decision, ensuring middleware runs before Docket.
+        Extracts task metadata from MCP request context and passes it explicitly
+        to call_tool(). The tool's _run() method handles the backgrounding decision,
+        ensuring middleware runs before Docket.
 
         Args:
             key: The name of the tool to call
@@ -1501,35 +1531,30 @@ async def _call_tool_mcp(
         Returns:
             Tool result or CreateTaskResult for background execution
         """
-        from fastmcp.server.dependencies import _docket_fn_key, _task_metadata
-
         logger.debug(
             f"[{self.name}] Handler called: call_tool %s with %s", key, arguments
         )
 
         try:
             # Extract SEP-1686 task metadata from request context
-            task_meta_dict: dict[str, Any] | None = None
+            task_meta: TaskMeta | None = None
             try:
                 ctx = self._mcp_server.request_context
                 if ctx.experimental.is_task:
-                    task_meta = ctx.experimental.task_metadata
-                    task_meta_dict = task_meta.model_dump(exclude_none=True)
+                    mcp_task_meta = ctx.experimental.task_metadata
+                    task_meta_dict = mcp_task_meta.model_dump(exclude_none=True)
+                    task_meta = TaskMeta(
+                        ttl=task_meta_dict.get("ttl"),
+                        fn_key=Tool.make_key(key),
+                    )
             except (AttributeError, LookupError):
                 pass
 
-            # Set contextvars so tool._run() can access them
-            task_token = _task_metadata.set(task_meta_dict)
-            key_token = _docket_fn_key.set(Tool.make_key(key))
-            try:
-                result = await self.call_tool(key, arguments)
+            result = await self.call_tool(key, arguments, task_meta=task_meta)
 
-                if isinstance(result, mcp.types.CreateTaskResult):
-                    return result
-                return result.to_mcp_result()
-            finally:
-                _task_metadata.reset(task_token)
-                _docket_fn_key.reset(key_token)
+            if isinstance(result, mcp.types.CreateTaskResult):
+                return result
+            return result.to_mcp_result()
 
         except DisabledError as e:
             raise NotFoundError(f"Unknown tool: {key!r}") from e
diff --git a/src/fastmcp/server/tasks/__init__.py b/src/fastmcp/server/tasks/__init__.py
index 10b59b66be..b3b4a72d4b 100644
--- a/src/fastmcp/server/tasks/__init__.py
+++ b/src/fastmcp/server/tasks/__init__.py
@@ -4,7 +4,7 @@
 """
 
 from fastmcp.server.tasks.capabilities import get_task_capabilities
-from fastmcp.server.tasks.config import TaskConfig, TaskMode
+from fastmcp.server.tasks.config import TaskConfig, TaskMeta, TaskMode
 from fastmcp.server.tasks.keys import (
     build_task_key,
     get_client_task_id_from_key,
@@ -13,6 +13,7 @@
 
 __all__ = [
     "TaskConfig",
+    "TaskMeta",
     "TaskMode",
     "build_task_key",
     "get_client_task_id_from_key",
diff --git a/src/fastmcp/server/tasks/config.py b/src/fastmcp/server/tasks/config.py
index 6531eba325..40da12ee85 100644
--- a/src/fastmcp/server/tasks/config.py
+++ b/src/fastmcp/server/tasks/config.py
@@ -21,6 +21,22 @@
 DEFAULT_TTL_MS = 60_000  # Default TTL in milliseconds
 
 
+@dataclass
+class TaskMeta:
+    """Metadata for task-augmented execution requests.
+
+    When passed to call_tool/read_resource/get_prompt, signals that
+    the operation should be submitted as a background task.
+
+    Attributes:
+        ttl: Client-requested TTL in milliseconds. If None, uses server default.
+        fn_key: Docket routing key. Auto-derived from component name if None.
+    """
+
+    ttl: int | None = None
+    fn_key: str | None = None
+
+
 @dataclass
 class TaskConfig:
     """Configuration for MCP background task execution (SEP-1686).
diff --git a/src/fastmcp/server/tasks/handlers.py b/src/fastmcp/server/tasks/handlers.py
index cf6573a5aa..45d30d92ca 100644
--- a/src/fastmcp/server/tasks/handlers.py
+++ b/src/fastmcp/server/tasks/handlers.py
@@ -15,6 +15,7 @@
 from mcp.types import INTERNAL_ERROR, ErrorData
 
 from fastmcp.server.dependencies import _current_docket, get_context
+from fastmcp.server.tasks.config import TaskMeta
 from fastmcp.server.tasks.keys import build_task_key
 
 if TYPE_CHECKING:
@@ -32,6 +33,7 @@ async def submit_to_docket(
     key: str,
     component: Tool | Resource | ResourceTemplate | Prompt,
     arguments: dict[str, Any] | None = None,
+    task_meta: TaskMeta | None = None,
 ) -> mcp.types.CreateTaskResult:
     """Submit any component to Docket for background execution (SEP-1686).
 
@@ -41,15 +43,13 @@ async def submit_to_docket(
     Queues the component's method to Docket, stores raw return values,
     and converts to MCP types on retrieval.
 
-    Note: Client-requested TTL in task_meta is intentionally ignored.
-    Server-side TTL policy (docket.execution_ttl) takes precedence for
-    consistent task lifecycle management.
-
     Args:
         task_type: Component type for task key construction
         key: The component key as seen by MCP layer (with namespace prefix)
         component: The component instance (Tool, Resource, ResourceTemplate, Prompt)
         arguments: Arguments/params (None for Resource which has no args)
+        task_meta: Task execution metadata. If task_meta.ttl is provided, it
+            overrides the server default (docket.execution_ttl).
 
     Returns:
         CreateTaskResult: Task stub with proper Task object
@@ -61,9 +61,12 @@ async def submit_to_docket(
     # Record creation timestamp per SEP-1686 final spec (line 430)
     created_at = datetime.now(timezone.utc)
 
-    # Get session ID and Docket
+    # Get session ID - use "internal" for programmatic calls without MCP session
     ctx = get_context()
-    session_id = ctx.session_id
+    try:
+        session_id = ctx.session_id
+    except RuntimeError:
+        session_id = "internal"
 
     docket = _current_docket.get()
     if docket is None:
@@ -77,13 +80,17 @@ async def submit_to_docket(
     # Build full task key with embedded metadata
     task_key = build_task_key(session_id, server_task_id, task_type, key)
 
+    # Determine TTL: use task_meta.ttl if provided, else docket default
+    if task_meta is not None and task_meta.ttl is not None:
+        ttl_ms = task_meta.ttl
+    else:
+        ttl_ms = int(docket.execution_ttl.total_seconds() * 1000)
+    ttl_seconds = int(ttl_ms / 1000) + TASK_MAPPING_TTL_BUFFER_SECONDS
+
     # Store task metadata in Redis for protocol handlers
     redis_key = f"fastmcp:task:{session_id}:{server_task_id}"
     created_at_key = f"fastmcp:task:{session_id}:{server_task_id}:created_at"
     poll_interval_key = f"fastmcp:task:{session_id}:{server_task_id}:poll_interval"
-    ttl_seconds = int(
-        docket.execution_ttl.total_seconds() + TASK_MAPPING_TTL_BUFFER_SECONDS
-    )
     poll_interval_ms = int(component.task_config.poll_interval.total_seconds() * 1000)
     async with docket.redis() as redis:
         await redis.set(redis_key, task_key, ex=ttl_seconds)
@@ -140,7 +147,7 @@ async def submit_to_docket(
             status="working",
             createdAt=created_at,
             lastUpdatedAt=created_at,
-            ttl=int(docket.execution_ttl.total_seconds() * 1000),
+            ttl=ttl_ms,
             pollInterval=poll_interval_ms,
         )
     )
diff --git a/src/fastmcp/server/tasks/routing.py b/src/fastmcp/server/tasks/routing.py
index 39f16b52a2..0f8fb5bc2a 100644
--- a/src/fastmcp/server/tasks/routing.py
+++ b/src/fastmcp/server/tasks/routing.py
@@ -12,6 +12,7 @@
 from mcp.types import METHOD_NOT_FOUND, ErrorData
 
 from fastmcp.server.dependencies import get_task_metadata
+from fastmcp.server.tasks.config import TaskMeta
 from fastmcp.server.tasks.handlers import submit_to_docket
 
 if TYPE_CHECKING:
@@ -26,16 +27,21 @@
 async def check_background_task(
     component: Tool | Resource | ResourceTemplate | Prompt,
     task_type: TaskType,
-    key: str,
+    # TODO: Remove `key` parameter when resources and prompts are updated to use
+    # explicit task_meta parameter like tools do
+    key: str | None = None,
     arguments: dict[str, Any] | None = None,
+    task_meta: TaskMeta | None = None,
 ) -> mcp.types.CreateTaskResult | None:
     """Check task mode and submit to background if requested.
 
     Args:
         component: The MCP component
         task_type: Type of task ("tool", "resource", "template", "prompt")
-        key: Docket registration key (caller resolves from contextvar + fallback)
+        key: Docket registration key (deprecated, use task_meta.fn_key instead)
         arguments: Arguments for tool/prompt/template execution
+        task_meta: Task execution metadata. If provided, execute as background task.
+            When None, falls back to reading from contextvar for backwards compat.
 
     Returns:
         CreateTaskResult if submitted to docket, None for sync execution
@@ -44,7 +50,16 @@ async def check_background_task(
         McpError: If mode="required" but no task metadata, or mode="forbidden"
                   but task metadata is present
     """
-    task_meta = get_task_metadata()
+    # For backwards compatibility: if task_meta not provided, check contextvar
+    # This is used by resources/prompts which haven't been updated yet
+    if task_meta is None:
+        task_meta_dict = get_task_metadata()
+        if task_meta_dict is not None:
+            task_meta = TaskMeta(
+                ttl=task_meta_dict.get("ttl"),
+                fn_key=key,  # Use key parameter for backwards compat
+            )
+
     task_config = component.task_config
 
     # Infer label from component
@@ -72,4 +87,7 @@ async def check_background_task(
     if not task_meta:
         return None
 
-    return await submit_to_docket(task_type, key, component, arguments)
+    # fn_key should be set by caller (FastMCP.call_tool enriches it)
+    # Fall back to key parameter for backwards compat, then component.key
+    fn_key = task_meta.fn_key or key or component.key
+    return await submit_to_docket(task_type, fn_key, component, arguments, task_meta)
diff --git a/src/fastmcp/tools/tool.py b/src/fastmcp/tools/tool.py
index ce897bc845..4c0d3aa3aa 100644
--- a/src/fastmcp/tools/tool.py
+++ b/src/fastmcp/tools/tool.py
@@ -31,7 +31,7 @@
 
 import fastmcp
 from fastmcp.server.dependencies import without_injected_parameters
-from fastmcp.server.tasks.config import TaskConfig
+from fastmcp.server.tasks.config import TaskConfig, TaskMeta
 from fastmcp.utilities.components import FastMCPComponent
 from fastmcp.utilities.json_schema import compress_schema, resolve_root_ref
 from fastmcp.utilities.logging import get_logger
@@ -278,7 +278,9 @@ def convert_result(self, raw_value: Any) -> ToolResult:
         )
 
     async def _run(
-        self, arguments: dict[str, Any]
+        self,
+        arguments: dict[str, Any],
+        task_meta: TaskMeta | None = None,
     ) -> ToolResult | mcp.types.CreateTaskResult:
         """Server entry point that handles task routing.
 
@@ -286,16 +288,22 @@ async def _run(
         task_config.mode to "supported" or "required". The server calls this
         method instead of run() directly.
 
+        Args:
+            arguments: Tool arguments
+            task_meta: If provided, execute as background task. If None, execute
+                synchronously.
+
         Subclasses can override this to customize task routing behavior.
         For example, FastMCPProviderTool overrides to delegate to child
         middleware without submitting to Docket.
         """
-        from fastmcp.server.dependencies import _docket_fn_key
         from fastmcp.server.tasks.routing import check_background_task
 
-        key = _docket_fn_key.get() or self.key
         task_result = await check_background_task(
-            component=self, task_type="tool", key=key, arguments=arguments
+            component=self,
+            task_type="tool",
+            arguments=arguments,
+            task_meta=task_meta,
         )
         if task_result:
             return task_result
diff --git a/tests/server/providers/test_local_provider_tools.py b/tests/server/providers/test_local_provider_tools.py
index 9cca754c42..4ff72c72b8 100644
--- a/tests/server/providers/test_local_provider_tools.py
+++ b/tests/server/providers/test_local_provider_tools.py
@@ -150,7 +150,6 @@ def string_tool() -> str:
             return "Hello, world!"
 
         result = await mcp.call_tool("string_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "Hello, world!"}
 
     async def test_bytes(self, tmp_path: Path):
@@ -161,7 +160,6 @@ def bytes_tool() -> bytes:
             return b"Hello, world!"
 
         result = await mcp.call_tool("bytes_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "Hello, world!"}
 
     async def test_uuid(self):
@@ -174,7 +172,6 @@ def uuid_tool() -> uuid.UUID:
             return test_uuid
 
         result = await mcp.call_tool("uuid_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": str(test_uuid)}
 
     async def test_path(self):
@@ -187,7 +184,6 @@ def path_tool() -> Path:
             return test_path
 
         result = await mcp.call_tool("path_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": str(test_path)}
 
     async def test_datetime(self):
@@ -200,7 +196,6 @@ def datetime_tool() -> datetime.datetime:
             return dt
 
         result = await mcp.call_tool("datetime_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": dt.isoformat()}
 
     async def test_image(self, tmp_path: Path):
@@ -214,7 +209,6 @@ def image_tool(path: str) -> Image:
         image_path.write_bytes(b"fake png data")
 
         result = await mcp.call_tool("image_tool", {"path": str(image_path)})
-        assert isinstance(result, ToolResult)
         assert result.structured_content is None
         assert isinstance(result.content, list)
         content = result.content[0]
@@ -235,7 +229,6 @@ def audio_tool(path: str) -> Audio:
         audio_path.write_bytes(b"fake wav data")
 
         result = await mcp.call_tool("audio_tool", {"path": str(audio_path)})
-        assert isinstance(result, ToolResult)
         assert isinstance(result.content, list)
         content = result.content[0]
         assert isinstance(content, AudioContent)
@@ -255,7 +248,6 @@ def file_tool(path: str) -> File:
         file_path.write_bytes(b"test file data")
 
         result = await mcp.call_tool("file_tool", {"path": str(file_path)})
-        assert isinstance(result, ToolResult)
         assert isinstance(result.content, list)
         content = result.content[0]
         assert isinstance(content, EmbeddedResource)
@@ -270,7 +262,6 @@ def file_tool(path: str) -> File:
 
     async def test_tool_mixed_content(self, tool_server: FastMCP):
         result = await tool_server.call_tool("mixed_content_tool", {})
-        assert isinstance(result, ToolResult)
         assert isinstance(result.content, list)
         assert len(result.content) == 3
         content1 = result.content[0]
@@ -301,7 +292,6 @@ async def test_tool_mixed_list_with_image(
         result = await tool_server.call_tool(
             "mixed_list_fn", {"image_path": str(image_path)}
         )
-        assert isinstance(result, ToolResult)
         assert isinstance(result.content, list)
         assert len(result.content) == 4
         content1 = result.content[0]
@@ -329,7 +319,6 @@ async def test_tool_mixed_list_with_audio(
         result = await tool_server.call_tool(
             "mixed_audio_list_fn", {"audio_path": str(audio_path)}
         )
-        assert isinstance(result, ToolResult)
         assert isinstance(result.content, list)
         assert len(result.content) == 4
         content1 = result.content[0]
@@ -357,7 +346,6 @@ async def test_tool_mixed_list_with_file(
         result = await tool_server.call_tool(
             "mixed_file_list_fn", {"file_path": str(file_path)}
         )
-        assert isinstance(result, ToolResult)
         assert isinstance(result.content, list)
         assert len(result.content) == 4
         content1 = result.content[0]
@@ -434,7 +422,6 @@ def process_image(image: bytes) -> Image:
             return Image(data=image)
 
         result = await mcp.call_tool("process_image", {"image": b"fake png data"})
-        assert isinstance(result, ToolResult)
         assert result.structured_content is None
         assert isinstance(result.content, list)
         assert isinstance(result.content[0], ImageContent)
@@ -465,7 +452,6 @@ def add_one(x: int) -> int:
             return x + 1
 
         result = await mcp.call_tool("add_one", {"x": "42"})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 43}
 
     async def test_tool_bool_coercion(self):
@@ -477,11 +463,9 @@ def toggle(flag: bool) -> bool:
             return not flag
 
         result = await mcp.call_tool("toggle", {"flag": "true"})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": False}
 
         result = await mcp.call_tool("toggle", {"flag": "false"})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": True}
 
     async def test_annotated_field_validation(self):
@@ -549,7 +533,6 @@ def analyze(x: Literal["a", "b"]) -> str:
             return x
 
         result = await mcp.call_tool("analyze", {"x": "a"})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "a"}
 
     async def test_enum_type_validation_error(self):
@@ -585,7 +568,6 @@ def analyze(x: MyEnum) -> str:
             return x.value
 
         result = await mcp.call_tool("analyze", {"x": "red"})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "red"}
 
     async def test_union_type_validation(self):
@@ -598,11 +580,9 @@ def analyze(x: int | float) -> str:
             return str(x)
 
         result = await mcp.call_tool("analyze", {"x": 1})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "1"}
 
         result = await mcp.call_tool("analyze", {"x": 1.0})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "1.0"}
 
         with pytest.raises(
@@ -622,7 +602,6 @@ def send_path(path: Path) -> str:
         test_path = Path("tmp") / "test.txt"
 
         result = await mcp.call_tool("send_path", {"path": str(test_path)})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": str(test_path)}
 
     async def test_path_type_error(self):
@@ -648,7 +627,6 @@ def send_uuid(x: uuid.UUID) -> str:
         test_uuid = uuid.uuid4()
 
         result = await mcp.call_tool("send_uuid", {"x": test_uuid})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": str(test_uuid)}
 
     async def test_uuid_type_error(self):
@@ -673,7 +651,6 @@ def send_datetime(x: datetime.datetime) -> str:
         dt = datetime.datetime(2025, 4, 25, 1, 2, 3)
 
         result = await mcp.call_tool("send_datetime", {"x": dt})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": dt.isoformat()}
 
     async def test_datetime_type_parse_string(self):
@@ -684,7 +661,6 @@ def send_datetime(x: datetime.datetime) -> str:
             return x.isoformat()
 
         result = await mcp.call_tool("send_datetime", {"x": "2021-01-01T00:00:00"})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "2021-01-01T00:00:00"}
 
     async def test_datetime_type_error(self):
@@ -707,7 +683,6 @@ def send_date(x: datetime.date) -> str:
             return x.isoformat()
 
         result = await mcp.call_tool("send_date", {"x": datetime.date.today()})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {
             "result": datetime.date.today().isoformat()
         }
@@ -720,7 +695,6 @@ def send_date(x: datetime.date) -> str:
             return x.isoformat()
 
         result = await mcp.call_tool("send_date", {"x": "2021-01-01"})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "2021-01-01"}
 
     async def test_timedelta_type(self):
@@ -733,7 +707,6 @@ def send_timedelta(x: datetime.timedelta) -> str:
         result = await mcp.call_tool(
             "send_timedelta", {"x": datetime.timedelta(days=1)}
         )
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "1 day, 0:00:00"}
 
     async def test_timedelta_type_parse_int(self):
@@ -745,7 +718,6 @@ def send_timedelta(x: datetime.timedelta) -> str:
             return str(x)
 
         result = await mcp.call_tool("send_timedelta", {"x": 1000})
-        assert isinstance(result, ToolResult)
         assert result.structured_content is not None
         result_str = result.structured_content["result"]
         assert (
@@ -815,7 +787,6 @@ def f() -> int:
             return 42
 
         result = await mcp.call_tool("f", {})
-        assert isinstance(result, ToolResult)
         assert isinstance(result.content, list)
         assert isinstance(result.content[0], TextContent)
         assert result.content[0].text == "42"
@@ -833,7 +804,6 @@ def f() -> ToolResult:
         assert f.output_schema is None
 
         result = await mcp.call_tool("f", {})
-        assert isinstance(result, ToolResult)
         assert isinstance(result.content, list)
         assert isinstance(result.content[0], TextContent)
         assert result.content[0].text == "Hello, world!"
@@ -852,7 +822,6 @@ def simple_tool() -> int:
         assert tool.output_schema is None
 
         result = await mcp.call_tool("simple_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content is None
         assert isinstance(result.content, list)
         assert isinstance(result.content[0], TextContent)
@@ -888,7 +857,6 @@ def explicit_tool() -> dict[str, Any]:
         assert tool.output_schema == expected_schema
 
         result = await mcp.call_tool("explicit_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"greeting": "Hello", "count": 42}
 
     async def test_output_schema_wrapped_primitive(self):
@@ -910,7 +878,6 @@ def primitive_tool() -> str:
         assert tool.output_schema == expected_schema
 
         result = await mcp.call_tool("primitive_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "Hello, primitives!"}
 
     async def test_output_schema_complex_type(self):
@@ -935,7 +902,6 @@ def complex_tool() -> list[dict[str, int]]:
         assert tool.output_schema == expected_schema
 
         result = await mcp.call_tool("complex_tool", {})
-        assert isinstance(result, ToolResult)
         expected_data = [{"a": 1, "b": 2}, {"c": 3, "d": 4}]
         assert result.structured_content == {"result": expected_data}
 
@@ -961,7 +927,6 @@ def dataclass_tool() -> User:
         assert tool.output_schema and "x-fastmcp-wrap-result" not in tool.output_schema
 
         result = await mcp.call_tool("dataclass_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"name": "Alice", "age": 30}
 
     async def test_output_schema_mixed_content_types(self):
@@ -977,7 +942,6 @@ def mixed_output() -> list[Any]:
             ]
 
         result = await mcp.call_tool("mixed_output", {})
-        assert isinstance(result, ToolResult)
         assert isinstance(result.content, list)
         assert len(result.content) == 3
         assert isinstance(result.content[0], TextContent)
@@ -1001,7 +965,6 @@ def edge_case_tool() -> tuple[int, str]:
         assert tool.output_schema and "x-fastmcp-wrap-result" in tool.output_schema
 
         result = await mcp.call_tool("edge_case_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": [42, "hello"]}
 
 
@@ -1032,7 +995,6 @@ def tool_with_context(x: int, ctx: Context) -> str:
             return f"Got context with x={x}"
 
         result = await mcp.call_tool("tool_with_context", {"x": 42})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "Got context with x=42"}
 
     async def test_async_context(self):
@@ -1045,7 +1007,6 @@ async def async_tool(x: int, ctx: Context) -> str:
             return f"Async with x={x}"
 
         result = await mcp.call_tool("async_tool", {"x": 42})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "Async with x=42"}
 
     async def test_optional_context(self):
@@ -1057,7 +1018,6 @@ def no_context(x: int) -> int:
             return x * 2
 
         result = await mcp.call_tool("no_context", {"x": 21})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 42}
 
     async def test_context_resource_access(self):
@@ -1076,7 +1036,6 @@ async def tool_with_resource(ctx: Context) -> str:
             return f"Read resource: {r.content} with mime type {r.mime_type}"
 
         result = await mcp.call_tool("tool_with_resource", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {
             "result": "Read resource: resource data with mime type text/plain"
         }
@@ -1105,7 +1064,6 @@ async def __call__(self, x: int, ctx: Context) -> int:
         mcp.add_tool(Tool.from_function(MyTool(), name="MyTool"))
 
         result = await mcp.call_tool("MyTool", {"x": 2})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 3}
 
     async def test_decorated_tool_with_functools_wraps(self):
@@ -1131,7 +1089,6 @@ async def decorated_tool(ctx: Context, query: str) -> str:
         assert "ctx" not in tool.parameters.get("properties", {})
 
         result = await mcp.call_tool("decorated_tool", {"query": "test"})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "query: test"}
 
 
@@ -1152,7 +1109,6 @@ def add(x: int, y: int) -> int:
             return x + y
 
         result = await mcp.call_tool("add", {"x": 1, "y": 2})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 3}
 
     async def test_tool_decorator_without_parentheses(self):
@@ -1167,7 +1123,6 @@ def add(x: int, y: int) -> int:
         assert any(t.name == "add" for t in tools)
 
         result = await mcp.call_tool("add", {"x": 1, "y": 2})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 3}
 
     async def test_tool_decorator_with_name(self):
@@ -1178,7 +1133,6 @@ def add(x: int, y: int) -> int:
             return x + y
 
         result = await mcp.call_tool("custom-add", {"x": 1, "y": 2})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 3}
 
     async def test_tool_decorator_with_description(self):
@@ -1206,7 +1160,6 @@ def add(self, y: int) -> int:
         obj = MyClass(10)
         mcp.add_tool(Tool.from_function(obj.add))
         result = await mcp.call_tool("add", {"y": 2})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 12}
 
     async def test_tool_decorator_classmethod(self):
@@ -1221,7 +1174,6 @@ def add(cls, y: int) -> int:
 
         mcp.add_tool(Tool.from_function(MyClass.add))
         result = await mcp.call_tool("add", {"y": 2})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 12}
 
     async def test_tool_decorator_staticmethod(self):
@@ -1234,7 +1186,6 @@ def add(x: int, y: int) -> int:
                 return x + y
 
         result = await mcp.call_tool("add", {"x": 1, "y": 2})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 3}
 
     async def test_tool_decorator_async_function(self):
@@ -1245,7 +1196,6 @@ async def add(x: int, y: int) -> int:
             return x + y
 
         result = await mcp.call_tool("add", {"x": 1, "y": 2})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 3}
 
     async def test_tool_decorator_classmethod_error(self):
@@ -1271,7 +1221,6 @@ async def add(cls, y: int) -> int:
 
         mcp.add_tool(Tool.from_function(MyClass.add))
         result = await mcp.call_tool("add", {"y": 2})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 12}
 
     async def test_tool_decorator_staticmethod_async_function(self):
@@ -1284,7 +1233,6 @@ async def add(x: int, y: int) -> int:
 
         mcp.add_tool(Tool.from_function(MyClass.add))
         result = await mcp.call_tool("add", {"x": 1, "y": 2})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 3}
 
     async def test_tool_decorator_staticmethod_order(self):
@@ -1298,7 +1246,6 @@ def add_v1(x: int, y: int) -> int:
                 return x + y
 
         result = await mcp.call_tool("add_v1", {"x": 1, "y": 2})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 3}
 
     async def test_tool_decorator_with_tags(self):
@@ -1327,7 +1274,6 @@ def multiply(a: int, b: int) -> int:
         assert any(t.name == "custom_multiply" for t in tools)
 
         result = await mcp.call_tool("custom_multiply", {"a": 5, "b": 3})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 15}
 
         assert not any(t.name == "multiply" for t in tools)
@@ -1383,7 +1329,6 @@ def standalone_function(x: int, y: int) -> int:
         assert tool is result_fn
 
         result = await mcp.call_tool("direct_call_tool", {"x": 5, "y": 3})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 8}
 
     async def test_tool_decorator_with_string_name(self):
@@ -1400,7 +1345,6 @@ def my_function(x: int) -> str:
         assert not any(t.name == "my_function" for t in tools)
 
         result = await mcp.call_tool("string_named_tool", {"x": 42})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "Result: 42"}
 
     async def test_tool_decorator_conflicting_names_error(self):
diff --git a/tests/server/tasks/test_task_meta_parameter.py b/tests/server/tasks/test_task_meta_parameter.py
new file mode 100644
index 0000000000..49c4ac371c
--- /dev/null
+++ b/tests/server/tasks/test_task_meta_parameter.py
@@ -0,0 +1,314 @@
+"""
+Tests for the explicit task_meta parameter on FastMCP.call_tool().
+
+These tests verify that the task_meta parameter provides explicit control
+over sync vs task execution, replacing implicit contextvar-based behavior.
+"""
+
+import mcp.types
+import pytest
+
+from fastmcp import FastMCP
+from fastmcp.client import Client
+from fastmcp.exceptions import ToolError
+from fastmcp.server.middleware import CallNext, Middleware, MiddlewareContext
+from fastmcp.server.tasks.config import TaskMeta
+from fastmcp.tools.tool import Tool, ToolResult
+
+
+class TestTaskMetaParameter:
+    """Tests for task_meta parameter on FastMCP.call_tool()."""
+
+    async def test_task_meta_none_returns_tool_result(self):
+        """With task_meta=None (default), call_tool returns ToolResult."""
+        server = FastMCP("test")
+
+        @server.tool
+        async def simple_tool(x: int) -> int:
+            return x * 2
+
+        result = await server.call_tool("simple_tool", {"x": 5})
+
+        first_content = result.content[0]
+        assert isinstance(first_content, mcp.types.TextContent)
+        assert first_content.text == "10"
+
+    async def test_task_meta_none_on_task_enabled_tool_still_returns_tool_result(self):
+        """Even for task=True tools, task_meta=None returns ToolResult synchronously."""
+        server = FastMCP("test")
+
+        @server.tool(task=True)
+        async def task_enabled_tool(x: int) -> int:
+            return x * 2
+
+        # Without task_meta, should execute synchronously
+        result = await server.call_tool("task_enabled_tool", {"x": 5})
+
+        first_content = result.content[0]
+        assert isinstance(first_content, mcp.types.TextContent)
+        assert first_content.text == "10"
+
+    async def test_task_meta_on_forbidden_tool_raises_error(self):
+        """Providing task_meta to a task=False tool raises ToolError."""
+        server = FastMCP("test")
+
+        @server.tool(task=False)
+        async def sync_only_tool(x: int) -> int:
+            return x * 2
+
+        # Error is raised before docket is needed (McpError wrapped as ToolError)
+        with pytest.raises(ToolError) as exc_info:
+            await server.call_tool("sync_only_tool", {"x": 5}, task_meta=TaskMeta())
+
+        assert "does not support task-augmented execution" in str(exc_info.value)
+
+    async def test_task_meta_fn_key_auto_populated_in_call_tool(self):
+        """fn_key is auto-populated from tool name in call_tool()."""
+        server = FastMCP("test")
+
+        @server.tool(task=True)
+        async def auto_key_tool() -> str:
+            return "done"
+
+        # Verify fn_key starts as None
+        task_meta = TaskMeta()
+        assert task_meta.fn_key is None
+
+        # call_tool enriches the task_meta before passing to _run
+        # We test this via the client integration path
+        async with Client(server) as client:
+            result = await client.call_tool("auto_key_tool", {}, task=True)
+            # Should succeed because fn_key was auto-populated
+            from fastmcp.client.tasks import ToolTask
+
+            assert isinstance(result, ToolTask)
+
+    async def test_task_meta_fn_key_enrichment_logic(self):
+        """Verify that fn_key enrichment uses Tool.make_key()."""
+        # Direct test of the enrichment logic
+        tool_name = "my_tool"
+        expected_key = Tool.make_key(tool_name)
+
+        assert expected_key == "tool:my_tool"
+
+
+class TestTaskMetaTTL:
+    """Tests for task_meta.ttl behavior."""
+
+    async def test_task_with_custom_ttl_creates_task(self):
+        """task_meta.ttl is passed through when creating tasks."""
+        server = FastMCP("test")
+
+        @server.tool(task=True)
+        async def ttl_tool() -> str:
+            return "done"
+
+        custom_ttl_ms = 30000  # 30 seconds
+
+        async with Client(server) as client:
+            # Use client.call_tool with task=True and ttl
+            task = await client.call_tool("ttl_tool", {}, task=True, ttl=custom_ttl_ms)
+
+            from fastmcp.client.tasks import ToolTask
+
+            assert isinstance(task, ToolTask)
+
+            # Verify task completes successfully
+            result = await task.result()
+            assert "done" in str(result)
+
+    async def test_task_without_ttl_uses_default(self):
+        """task_meta.ttl=None uses docket.execution_ttl default."""
+        server = FastMCP("test")
+
+        @server.tool(task=True)
+        async def default_ttl_tool() -> str:
+            return "done"
+
+        async with Client(server) as client:
+            # Use client.call_tool with task=True, default ttl
+            task = await client.call_tool("default_ttl_tool", {}, task=True)
+
+            from fastmcp.client.tasks import ToolTask
+
+            assert isinstance(task, ToolTask)
+
+            # Verify task completes successfully
+            result = await task.result()
+            assert "done" in str(result)
+
+
+class TrackingMiddleware(Middleware):
+    """Middleware that tracks tool calls."""
+
+    def __init__(self, calls: list[str]):
+        super().__init__()
+        self._calls = calls
+
+    async def on_call_tool(
+        self,
+        context: MiddlewareContext[mcp.types.CallToolRequestParams],
+        call_next: CallNext[mcp.types.CallToolRequestParams, ToolResult],
+    ) -> ToolResult:
+        if context.method:
+            self._calls.append(context.method)
+        return await call_next(context)  # type: ignore[return-value]
+
+
+class TestTaskMetaMiddleware:
+    """Tests that task_meta is properly propagated through middleware."""
+
+    async def test_task_meta_propagated_through_middleware(self):
+        """task_meta is passed through middleware chain."""
+        server = FastMCP("test")
+        middleware_saw_request: list[str] = []
+
+        @server.tool(task=True)
+        async def middleware_test_tool() -> str:
+            return "done"
+
+        server.add_middleware(TrackingMiddleware(middleware_saw_request))
+
+        async with Client(server) as client:
+            # Use client to trigger the middleware chain
+            task = await client.call_tool("middleware_test_tool", {}, task=True)
+
+            # Middleware should have run
+            assert "tools/call" in middleware_saw_request
+
+            # And task should have been created
+            from fastmcp.client.tasks import ToolTask
+
+            assert isinstance(task, ToolTask)
+
+
+class TestTaskMetaClientIntegration:
+    """Tests that task_meta works correctly with the Client."""
+
+    async def test_client_task_true_maps_to_task_meta(self):
+        """Client's task=True creates proper task_meta on server."""
+        server = FastMCP("test")
+
+        @server.tool(task=True)
+        async def client_test_tool(x: int) -> int:
+            return x * 2
+
+        async with Client(server) as client:
+            # Client passes task=True, server receives as task_meta
+            task = await client.call_tool("client_test_tool", {"x": 5}, task=True)
+
+            # Should get back a ToolTask (client wrapper)
+            from fastmcp.client.tasks import ToolTask
+
+            assert isinstance(task, ToolTask)
+
+            # Wait for result
+            result = await task.result()
+            assert "10" in str(result)
+
+    async def test_client_without_task_gets_immediate_result(self):
+        """Client without task=True gets immediate result."""
+        server = FastMCP("test")
+
+        @server.tool(task=True)
+        async def immediate_tool(x: int) -> int:
+            return x * 2
+
+        async with Client(server) as client:
+            # No task=True, should execute synchronously
+            result = await client.call_tool("immediate_tool", {"x": 5})
+
+            # Should get CallToolResult directly
+            assert "10" in str(result)
+
+    async def test_client_task_with_custom_ttl(self):
+        """Client can pass custom TTL for task execution."""
+        server = FastMCP("test")
+
+        @server.tool(task=True)
+        async def custom_ttl_tool() -> str:
+            return "done"
+
+        custom_ttl_ms = 60000  # 60 seconds
+
+        async with Client(server) as client:
+            task = await client.call_tool(
+                "custom_ttl_tool", {}, task=True, ttl=custom_ttl_ms
+            )
+
+            from fastmcp.client.tasks import ToolTask
+
+            assert isinstance(task, ToolTask)
+
+            # Verify task completes successfully
+            result = await task.result()
+            assert "done" in str(result)
+
+
+class TestTaskMetaDirectServerCall:
+    """Tests for direct server calls (tool calling another tool)."""
+
+    async def test_tool_can_call_another_tool_with_task(self):
+        """A tool can call another tool as a background task."""
+        server = FastMCP("test")
+
+        @server.tool(task=True)
+        async def inner_tool(x: int) -> int:
+            return x * 2
+
+        @server.tool
+        async def outer_tool(x: int) -> str:
+            # Call inner tool as background task
+            result = await server.call_tool(
+                "inner_tool", {"x": x}, task_meta=TaskMeta()
+            )
+            # Should get CreateTaskResult since we're in server context
+            return f"Created task: {result.task.taskId}"
+
+        async with Client(server) as client:
+            # Call outer_tool which internally calls inner_tool with task_meta
+            result = await client.call_tool("outer_tool", {"x": 5})
+            # The outer tool should have successfully created a background task
+            assert "Created task:" in str(result)
+
+    async def test_tool_can_call_another_tool_synchronously(self):
+        """A tool can call another tool synchronously (no task_meta)."""
+        server = FastMCP("test")
+
+        @server.tool(task=True)
+        async def inner_tool(x: int) -> int:
+            return x * 2
+
+        @server.tool
+        async def outer_tool(x: int) -> str:
+            # Call inner tool synchronously (no task_meta)
+            result = await server.call_tool("inner_tool", {"x": x})
+            # Should get ToolResult directly
+            first_content = result.content[0]
+            assert isinstance(first_content, mcp.types.TextContent)
+            return f"Got result: {first_content.text}"
+
+        async with Client(server) as client:
+            result = await client.call_tool("outer_tool", {"x": 5})
+            assert "Got result: 10" in str(result)
+
+    async def test_tool_can_call_another_tool_with_custom_ttl(self):
+        """A tool can call another tool as a background task with custom TTL."""
+        server = FastMCP("test")
+
+        @server.tool(task=True)
+        async def inner_tool(x: int) -> int:
+            return x * 2
+
+        @server.tool
+        async def outer_tool(x: int) -> str:
+            custom_ttl = 45000  # 45 seconds
+            result = await server.call_tool(
+                "inner_tool", {"x": x}, task_meta=TaskMeta(ttl=custom_ttl)
+            )
+            return f"Task TTL: {result.task.ttl}"
+
+        async with Client(server) as client:
+            result = await client.call_tool("outer_tool", {"x": 5})
+            # The inner tool task should have the custom TTL
+            assert "Task TTL: 45000" in str(result)
diff --git a/tests/server/test_dependencies.py b/tests/server/test_dependencies.py
index 38676d51ad..37e2836304 100644
--- a/tests/server/test_dependencies.py
+++ b/tests/server/test_dependencies.py
@@ -11,7 +11,6 @@
 from fastmcp.prompts import PromptResult
 from fastmcp.resources import ResourceResult
 from fastmcp.server.context import Context
-from fastmcp.tools.tool import ToolResult
 
 HUZZAH = "huzzah!"
 
@@ -56,7 +55,6 @@ def fetch_data(query: str, config: dict[str, str] = Depends(get_config)) -> str:
         )
 
     result = await mcp.call_tool("fetch_data", {"query": "users"})
-    assert isinstance(result, ToolResult)
     assert result.structured_content is not None
     text = result.structured_content["result"]
     assert "Fetching 'users' from https://api.example.com" in text
@@ -74,7 +72,6 @@ async def greet_user(name: str, user_id: int = Depends(get_user_id)) -> str:  #
         return f"Hello {name}, your ID is {user_id}"
 
     result = await mcp.call_tool("greet_user", {"name": "Alice"})
-    assert isinstance(result, ToolResult)
     assert result.structured_content is not None
     assert result.structured_content["result"] == "Hello Alice, your ID is 42"
 
@@ -97,7 +94,6 @@ async def query_db(sql: str, db: str = Depends(get_database)) -> str:  # type: i
         return f"Executing '{sql}' on {db}"
 
     result = await mcp.call_tool("query_db", {"sql": "SELECT * FROM users"})
-    assert isinstance(result, ToolResult)
     assert result.structured_content is not None
     assert (
         "Executing 'SELECT * FROM users' on db_connection"
@@ -122,7 +118,6 @@ async def call_api(
         return f"Calling {client['base_url']}/{client['version']}/{endpoint}"
 
     result = await mcp.call_tool("call_api", {"endpoint": "users"})
-    assert isinstance(result, ToolResult)
     assert result.structured_content is not None
     assert (
         result.structured_content["result"]
@@ -160,7 +155,6 @@ def use_context(ctx: Context = CurrentContext()) -> str:
         return HUZZAH
 
     result = await mcp.call_tool("use_context", {})
-    assert isinstance(result, ToolResult)
     assert result.structured_content is not None
     assert result.structured_content["result"] == HUZZAH
 
@@ -179,7 +173,6 @@ def use_both_contexts(
         return HUZZAH
 
     result = await mcp.call_tool("use_both_contexts", {})
-    assert isinstance(result, ToolResult)
     assert result.structured_content is not None
     assert result.structured_content["result"] == HUZZAH
 
@@ -210,7 +203,6 @@ def process_data(value: int, config: str = Depends(fetch_config)) -> str:  # typ
         return f"Processing {value} with {config}"
 
     result = await mcp.call_tool("process_data", {"value": 100})
-    assert isinstance(result, ToolResult)
     assert result.structured_content is not None
     assert result.structured_content["result"] == "Processing 100 with loaded_config"
 
@@ -232,7 +224,6 @@ async def tool_with_cached_dep(
         return f"{dep1} + {dep2} = {dep1 + dep2}"
 
     result = await mcp.call_tool("tool_with_cached_dep", {})
-    assert isinstance(result, ToolResult)
     assert result.structured_content is not None
     assert result.structured_content["result"] == "42 + 42 = 84"
     assert call_count == 1
@@ -395,7 +386,6 @@ async def query_data(
         return f"open={connection.is_open}"
 
     result = await mcp.call_tool("query_data", {"query": "test"})
-    assert isinstance(result, ToolResult)
     assert result.structured_content is not None
     assert result.structured_content["result"] == "open=True"
 
@@ -464,7 +454,6 @@ async def validated_tool(
 
     # Valid argument
     result = await mcp.call_tool("validated_tool", {"age": 25})
-    assert isinstance(result, ToolResult)
     assert result.structured_content is not None
     assert result.structured_content["result"] == "age=25"
 
@@ -513,7 +502,6 @@ async def query_sync(
         return f"open={connection.is_open}"
 
     result = await mcp.call_tool("query_sync", {"query": "test"})
-    assert isinstance(result, ToolResult)
     assert result.structured_content is not None
     assert result.structured_content["result"] == "open=True"
     assert not conn.is_open
@@ -617,7 +605,6 @@ async def check_permission(
 
     # Normal call - dependency is resolved
     result = await mcp.call_tool("check_permission", {"action": "read"})
-    assert isinstance(result, ToolResult)
     assert result.structured_content is not None
     assert "admin=not_admin" in result.structured_content["result"]
 
diff --git a/tests/server/test_mount.py b/tests/server/test_mount.py
index 4470d7885d..ad20c2bdda 100644
--- a/tests/server/test_mount.py
+++ b/tests/server/test_mount.py
@@ -13,7 +13,7 @@
 from fastmcp.resources import ResourceResult
 from fastmcp.server.providers import FastMCPProvider, TransformingProvider
 from fastmcp.server.providers.proxy import FastMCPProxy
-from fastmcp.tools.tool import Tool, ToolResult
+from fastmcp.tools.tool import Tool
 from fastmcp.tools.tool_transform import TransformedTool
 from fastmcp.utilities.tests import caplog_for_fastmcp
 
@@ -47,7 +47,6 @@ def tool() -> str:
         assert any(t.name == "sub_transformed_tool" for t in tools)
 
         result = await main_app.call_tool("sub_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "This is from the sub app"}
 
     async def test_mount_with_custom_separator(self):
@@ -68,7 +67,6 @@ def greet(name: str) -> str:
 
         # Call the tool
         result = await main_app.call_tool("sub_greet", {"name": "World"})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "Hello, World!"}
 
     @pytest.mark.parametrize("prefix", ["", None])
@@ -105,7 +103,6 @@ def sub_tool() -> str:
 
         # Call the tool to verify it works
         result = await main_app.call_tool("sub_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "This is from the sub app"}
 
     async def test_mount_tools_no_prefix(self):
@@ -126,7 +123,6 @@ def sub_tool() -> str:
 
         # Test actual functionality
         tool_result = await main_app.call_tool("sub_tool", {})
-        assert isinstance(tool_result, ToolResult)
         assert tool_result.structured_content == {"result": "Sub tool result"}
 
     async def test_mount_resources_no_prefix(self):
@@ -221,10 +217,8 @@ def get_headlines() -> str:
 
         # Call tools from both mounted servers
         result1 = await main_app.call_tool("weather_get_forecast", {})
-        assert isinstance(result1, ToolResult)
         assert result1.structured_content == {"result": "Weather forecast"}
         result2 = await main_app.call_tool("news_get_headlines", {})
-        assert isinstance(result2, ToolResult)
         assert result2.structured_content == {"result": "News headlines"}
 
     async def test_mount_same_prefix(self):
@@ -359,7 +353,6 @@ def second_shared_tool() -> str:
 
         # Test that calling the tool uses the first server's implementation
         result = await main_app.call_tool("shared_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "First app tool"}
 
     async def test_first_server_wins_tools_same_prefix(self):
@@ -388,7 +381,6 @@ def second_shared_tool() -> str:
 
         # Test that calling the tool uses the first server's implementation
         result = await main_app.call_tool("api_shared_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "First app tool"}
 
     async def test_first_server_wins_resources_no_prefix(self):
@@ -600,7 +592,6 @@ def dynamic_tool() -> str:
 
         # Call the dynamically added tool
         result = await main_app.call_tool("sub_dynamic_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "Added after mounting"}
 
     async def test_removing_tool_after_mounting(self):
@@ -777,7 +768,6 @@ def get_data(query: str) -> str:
 
         # Call the tool
         result = await main_app.call_tool("proxy_get_data", {"query": "test"})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "Data for test"}
 
     async def test_dynamically_adding_to_proxied_server(self):
@@ -803,7 +793,6 @@ def dynamic_data() -> str:
 
         # Call the tool
         result = await main_app.call_tool("proxy_dynamic_data", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "Dynamic data"}
 
     async def test_proxy_server_with_resources(self):
@@ -1079,7 +1068,6 @@ def blocked_tool() -> str:
 
         # Verify execution also respects filters
         result = await parent.call_tool("allowed_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "allowed"}
 
         with pytest.raises(NotFoundError, match="Unknown tool"):
@@ -1251,12 +1239,10 @@ def multiply(a: int, b: int) -> int:
 
         # Tool at level 2 should work
         result = await root.call_tool("middle_multiply", {"a": 3, "b": 4})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 12}
 
         # Tool at level 3 should also work (this was the bug)
         result = await root.call_tool("middle_leaf_add", {"a": 5, "b": 7})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": 12}
 
     async def test_three_level_nested_resource_invocation(self):
@@ -1364,7 +1350,6 @@ def deep_tool() -> str:
 
         # Tool at level 4 should work
         result = await root.call_tool("l1_l2_l3_deep_tool", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "very deep"}
 
 
@@ -1435,7 +1420,6 @@ def original_tool() -> str:
         )
 
         result = await main.call_tool("renamed", {})
-        assert isinstance(result, ToolResult)
         assert result.structured_content == {"result": "success"}
 
     def test_duplicate_tool_rename_targets_raises_error(self):
diff --git a/tests/server/test_providers.py b/tests/server/test_providers.py
index 70a2973c09..24e4679f85 100644
--- a/tests/server/test_providers.py
+++ b/tests/server/test_providers.py
@@ -162,7 +162,6 @@ async def test_call_dynamic_tool(
             name="dynamic_multiply", arguments={"a": 7, "b": 6}
         )
 
-        assert isinstance(result, ToolResult)
         assert result.structured_content is not None
         assert isinstance(result.structured_content, dict)
         assert result.structured_content["result"] == 42
@@ -179,7 +178,6 @@ async def test_call_dynamic_tool_with_config(
             name="dynamic_add", arguments={"a": 5, "b": 3}
         )
 
-        assert isinstance(result, ToolResult)
         assert result.structured_content is not None
         # 5 + 3 + 100 (value offset) = 108
         assert isinstance(result.structured_content, dict)
@@ -196,7 +194,6 @@ async def test_call_static_tool_still_works(
             name="static_add", arguments={"a": 10, "b": 5}
         )
 
-        assert isinstance(result, ToolResult)
         assert result.structured_content is not None
         assert isinstance(result.structured_content, dict)
         assert result.structured_content["result"] == 15
@@ -232,7 +229,6 @@ async def test_default_get_tool_falls_back_to_list(self, base_server: FastMCP):
             name="test_tool", arguments={"a": 1, "b": 2}
         )
 
-        assert isinstance(result, ToolResult)
         assert result.structured_content is not None
         # Default get_tool should have called list_tools
         assert provider.list_tools_call_count >= 1
@@ -272,7 +268,6 @@ async def test_tool_not_found_falls_through_to_static(
             name="static_subtract", arguments={"a": 10, "b": 3}
         )
 
-        assert isinstance(result, ToolResult)
         assert result.structured_content is not None
         assert isinstance(result.structured_content, dict)
         assert result.structured_content["result"] == 7
@@ -371,7 +366,6 @@ async def test_call_tool_default_implementation(self):
 
         result = await mcp.call_tool("test_tool", {"a": 1, "b": 2})
 
-        assert isinstance(result, ToolResult)
         assert result.structured_content is not None
         assert isinstance(result.structured_content, dict)
         assert result.structured_content["result"] == 3