Trace client task management requests (#4525)

docs/servers/telemetry.mdx

@@ -6,7 +6,7 @@ icon: chart-line
 tag: NEW
 ---
 
-FastMCP includes native OpenTelemetry instrumentation for observability. Traces are automatically generated for tool, prompt, resource, and resource template operations, providing visibility into server behavior, request handling, and provider delegation chains.
+FastMCP includes native OpenTelemetry instrumentation for observability. Traces are automatically generated for tool, prompt, resource, resource template, and task management operations, providing visibility into server behavior, request handling, and provider delegation chains.
 
 ## How It Works
 
@@ -69,12 +69,13 @@ The server creates spans for each operation using [MCP semantic conventions](htt
 | `tools/call {name}` | Tool execution (e.g., `tools/call get_weather`) |
 | `resources/read` | Resource read (URI in `mcp.resource.uri` attribute, not span name) |
 | `prompts/get {name}` | Prompt render (e.g., `prompts/get greeting`) |
+| `tasks/{operation}` | Task management (`tasks/get`, `tasks/result`, `tasks/list`, or `tasks/cancel`) |
 
 For mounted servers, an additional `delegate {name}` span shows the delegation to the child server.
 
 ### Client Spans
 
-The FastMCP client creates spans for outgoing requests with the same naming pattern (`tools/call {name}`, `resources/read`, `prompts/get {name}`).
+The FastMCP client creates spans for outgoing requests with the same naming pattern (`tools/call {name}`, `resources/read`, `prompts/get {name}`, and `tasks/{operation}`).
 
 ### Span Hierarchy
 
@@ -95,6 +96,54 @@ tools/call remote_search (CLIENT)
         └── [remote server spans via trace context propagation]
 ```
 
+### Background tasks
+
+Background task traces have two parts:
+
+- Task submission and management requests use normal client-to-server context propagation. `tasks/get`, `tasks/result`, `tasks/list`, and `tasks/cancel` server spans are descendants of the corresponding FastMCP client spans.
+- Deferred execution runs in a Docket worker. Docket records its `CONSUMER` span as a new trace root with a span link to the submission context, rather than making it a child of the submission span. Custom spans created inside the task are children of that worker span.
+
+Span links preserve the causal relationship without forcing worker sampling to inherit the submit trace's sampling decision. Some tracing backends do not display links prominently, so the worker trace may look disconnected even though the link is present.
+
+Frequent status and list polling can produce more detail than you need. You can drop those client and server spans with a sampler that checks the span name before delegating to `ParentBased`:
+
+```python
+from opentelemetry import trace
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.sampling import (
+    ALWAYS_ON,
+    Decision,
+    ParentBased,
+    Sampler,
+    SamplingResult,
+)
+
+
+class DropTaskPolls(Sampler):
+    def __init__(self):
+        self._delegate = ParentBased(ALWAYS_ON)
+
+    def should_sample(self, parent_context, trace_id, name, *args, **kwargs):
+        if name in {"tasks/get", "tasks/list"}:
+            return SamplingResult(Decision.DROP)
+        return self._delegate.should_sample(
+            parent_context,
+            trace_id,
+            name,
+            *args,
+            **kwargs,
+        )
+
+    def get_description(self):
+        return "DropTaskPolls"
+
+
+provider = TracerProvider(sampler=DropTaskPolls())
+trace.set_tracer_provider(provider)
+```
+
+The name check must happen before `ParentBased` delegates. If the name-based sampler is nested inside `ParentBased`, it is not consulted for child spans whose parent was already sampled.
+
 ## Programmatic Configuration
 
 For more control, configure the SDK in your Python code before importing FastMCP:
@@ -276,7 +325,7 @@ FastMCP implements the [OpenTelemetry MCP semantic conventions](https://opentele
 
 | Attribute | Description |
 |-----------|-------------|
-| `mcp.method.name` | The MCP method being called (`tools/call`, `resources/read`, `prompts/get`) |
+| `mcp.method.name` | The MCP method being called (`tools/call`, `resources/read`, `prompts/get`, `tasks/get`, etc.) |
 | `mcp.protocol.version` | The negotiated MCP protocol version for the request |
 | `mcp.session.id` | Session identifier for the MCP connection |
 | `mcp.resource.uri` | The resource URI (for resource operations) |

fastmcp_slim/fastmcp/client/mixins/task_management.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
 import mcp_types
 from mcp import MCPError
@@ -23,6 +23,8 @@ from mcp_types import (
     PaginatedRequestParams,
 )
 
+from fastmcp.client.telemetry import client_span
+from fastmcp.telemetry import inject_trace_context
 from fastmcp.utilities.logging import get_logger
 
 logger = get_logger(__name__)
@@ -64,13 +66,27 @@ class ClientTaskManagementMixin:
             RuntimeError: If client not connected
             MCPError: If the request results in a TimeoutError | JSONRPCError
         """
-        request = GetTaskRequest(params=GetTaskRequestParams(task_id=task_id))
-        return await self._await_with_session_monitoring(
-            self.session.send_request(
-                request=request,  # type: ignore[arg-type]
-                result_type=GetTaskResult,
+        with client_span(
+            "tasks/get",
+            "tasks/get",
+            task_id,
+            session_id=self.transport.get_session_id(),
+        ):
+            request_meta = cast(
+                "mcp_types.RequestParamsMeta | None", inject_trace_context()
+            )
+            request = GetTaskRequest(
+                params=GetTaskRequestParams(
+                    task_id=task_id,
+                    _meta=request_meta,  # type: ignore[unknown-argument]
+                )
+            )
+            return await self._await_with_session_monitoring(
+                self.session.send_request(
+                    request=request,  # type: ignore[arg-type]
+                    result_type=GetTaskResult,
+                )
             )
-        )
 
     async def get_task_result(self: Client, task_id: str) -> Any:
         """Retrieve the raw result of a completed background task.
@@ -88,20 +104,32 @@ class ClientTaskManagementMixin:
             RuntimeError: If client not connected, task not found, or task failed
             MCPError: If the request results in a TimeoutError | JSONRPCError
         """
-        request = GetTaskPayloadRequest(
-            params=GetTaskPayloadRequestParams(task_id=task_id)
-        )
-        # Return raw result - Task classes handle type-specific parsing
-        result = await self._await_with_session_monitoring(
-            self.session.send_request(
-                request=request,  # type: ignore[arg-type]
-                result_type=_RawTaskPayloadResult,
+        with client_span(
+            "tasks/result",
+            "tasks/result",
+            task_id,
+            session_id=self.transport.get_session_id(),
+        ):
+            request_meta = cast(
+                "mcp_types.RequestParamsMeta | None", inject_trace_context()
             )
-        )
-        # Return as dict for compatibility with Task class parsing. The payload
-        # fields (content, structuredContent, messages, contents, ...) survive
-        # via the permissive result type's extra="allow".
-        return result.model_dump(exclude_none=True, by_alias=True)
+            request = GetTaskPayloadRequest(
+                params=GetTaskPayloadRequestParams(
+                    task_id=task_id,
+                    _meta=request_meta,  # type: ignore[unknown-argument]
+                )
+            )
+            # Return raw result - Task classes handle type-specific parsing
+            result = await self._await_with_session_monitoring(
+                self.session.send_request(
+                    request=request,  # type: ignore[arg-type]
+                    result_type=_RawTaskPayloadResult,
+                )
+            )
+            # Return as dict for compatibility with Task class parsing. The payload
+            # fields (content, structuredContent, messages, contents, ...) survive
+            # via the permissive result type's extra="allow".
+            return result.model_dump(exclude_none=True, by_alias=True)
 
     async def list_tasks(
         self: Client,
@@ -127,31 +155,45 @@ class ClientTaskManagementMixin:
             RuntimeError: If client not connected
             MCPError: If the request results in a TimeoutError | JSONRPCError
         """
-        # Send protocol request
-        params = PaginatedRequestParams(cursor=cursor, limit=limit)  # type: ignore[call-arg]  # Optional field in MCP SDK  # ty:ignore[unknown-argument]
-        request = ListTasksRequest(params=params)
-        server_response = await self._await_with_session_monitoring(
-            self.session.send_request(
-                request=request,  # type: ignore[invalid-argument-type]
-                result_type=mcp_types.ListTasksResult,
+        with client_span(
+            "tasks/list",
+            "tasks/list",
+            "",
+            session_id=self.transport.get_session_id(),
+        ):
+            request_meta = cast(
+                "mcp_types.RequestParamsMeta | None", inject_trace_context()
             )
-        )
 
-        # If server returned tasks, use those
-        if server_response.tasks:
-            return server_response.model_dump(by_alias=True)
+            # Send protocol request
+            params = PaginatedRequestParams(
+                cursor=cursor,
+                limit=limit,  # type: ignore[call-arg]  # Optional field in MCP SDK  # ty:ignore[unknown-argument]
+                _meta=request_meta,  # type: ignore[unknown-argument]
+            )
+            request = ListTasksRequest(params=params)
+            server_response = await self._await_with_session_monitoring(
+                self.session.send_request(
+                    request=request,  # type: ignore[invalid-argument-type]
+                    result_type=mcp_types.ListTasksResult,
+                )
+            )
 
-        # Server returned empty - fall back to client-side tracking
-        tasks = []
-        for task_id in list(self._submitted_task_ids)[:limit]:
-            try:
-                status = await self.get_task_status(task_id)
-                tasks.append(status.model_dump(by_alias=True))
-            except MCPError:
-                # Task may have expired or been deleted, skip it
-                continue
+            # If server returned tasks, use those
+            if server_response.tasks:
+                return server_response.model_dump(by_alias=True)
 
-        return {"tasks": tasks, "nextCursor": None}
+            # Server returned empty - fall back to client-side tracking
+            tasks = []
+            for task_id in list(self._submitted_task_ids)[:limit]:
+                try:
+                    status = await self.get_task_status(task_id)
+                    tasks.append(status.model_dump(by_alias=True))
+                except MCPError:
+                    # Task may have expired or been deleted, skip it
+                    continue
+
+            return {"tasks": tasks, "nextCursor": None}
 
     async def cancel_task(self: Client, task_id: str) -> mcp_types.CancelTaskResult:
         """Cancel a task, transitioning it to cancelled state.
@@ -169,10 +211,24 @@ class ClientTaskManagementMixin:
             RuntimeError: If task doesn't exist
             MCPError: If the request results in a TimeoutError | JSONRPCError
         """
-        request = CancelTaskRequest(params=CancelTaskRequestParams(task_id=task_id))
-        return await self._await_with_session_monitoring(
-            self.session.send_request(
-                request=request,  # type: ignore[invalid-argument-type]
-                result_type=mcp_types.CancelTaskResult,
+        with client_span(
+            "tasks/cancel",
+            "tasks/cancel",
+            task_id,
+            session_id=self.transport.get_session_id(),
+        ):
+            request_meta = cast(
+                "mcp_types.RequestParamsMeta | None", inject_trace_context()
+            )
+            request = CancelTaskRequest(
+                params=CancelTaskRequestParams(
+                    task_id=task_id,
+                    _meta=request_meta,  # type: ignore[unknown-argument]
+                )
+            )
+            return await self._await_with_session_monitoring(
+                self.session.send_request(
+                    request=request,  # type: ignore[invalid-argument-type]
+                    result_type=mcp_types.CancelTaskResult,
+                )
             )
-        )

tests/client/telemetry/test_client_task_tracing.py

@@ -0,0 +1,93 @@
+"""Tests for client OpenTelemetry tracing on task operations."""
+
+import asyncio
+
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
+    InMemorySpanExporter,
+)
+from opentelemetry.trace import SpanKind
+
+from fastmcp import Client, FastMCP
+
+
+def assert_propagating_client_span(
+    trace_exporter: InMemorySpanExporter,
+    method: str,
+    component_key: str,
+) -> None:
+    all_spans = trace_exporter.get_finished_spans()
+    spans = [span for span in all_spans if span.name == method]
+    client_span = next(
+        span
+        for span in spans
+        if span.attributes is not None and "fastmcp.server.name" not in span.attributes
+    )
+    server_span = next(
+        span
+        for span in spans
+        if span.attributes is not None and "fastmcp.server.name" in span.attributes
+    )
+
+    assert client_span.kind == SpanKind.CLIENT
+    assert client_span.attributes is not None
+    assert client_span.attributes["mcp.method.name"] == method
+    assert client_span.attributes["fastmcp.component.key"] == component_key
+    assert server_span.parent is not None
+    assert server_span.context.trace_id == client_span.context.trace_id
+
+    spans_by_id = {span.context.span_id: span for span in all_spans}
+    current = server_span
+    while current.parent is not None:
+        parent = spans_by_id.get(current.parent.span_id)
+        assert parent is not None
+        if parent.context.span_id == client_span.context.span_id:
+            break
+        current = parent
+    else:
+        raise AssertionError("Server span should descend from the client span")
+
+
+async def test_list_tasks_creates_propagating_client_span(
+    trace_exporter: InMemorySpanExporter,
+):
+    server = FastMCP("test-server")
+
+    async with Client(server) as client:
+        await client.list_tasks()
+
+    assert_propagating_client_span(trace_exporter, "tasks/list", "")
+
+
+async def test_task_id_operations_create_propagating_client_spans(
+    trace_exporter: InMemorySpanExporter,
+):
+    started = asyncio.Event()
+    server = FastMCP("test-server")
+
+    @server.tool(task=True)
+    async def quick_tool() -> str:
+        return "done"
+
+    @server.tool(task=True)
+    async def slow_tool() -> str:
+        started.set()
+        await asyncio.sleep(10)
+        return "done"
+
+    async with Client(server) as client:
+        completed_task = await client.call_tool("quick_tool", task=True)
+        await completed_task.wait(timeout=2)
+        trace_exporter.clear()
+
+        await client.get_task_status(completed_task.task_id)
+        await client.get_task_result(completed_task.task_id)
+
+        running_task = await client.call_tool("slow_tool", task=True)
+        await asyncio.wait_for(started.wait(), timeout=2)
+        await client.cancel_task(running_task.task_id)
+
+    assert_propagating_client_span(trace_exporter, "tasks/get", completed_task.task_id)
+    assert_propagating_client_span(
+        trace_exporter, "tasks/result", completed_task.task_id
+    )
+    assert_propagating_client_span(trace_exporter, "tasks/cancel", running_task.task_id)