Skip to content

task_handlers

Experimental task handler protocols for server -> client requests.

This module provides Protocol types and default handlers for when servers send task-related requests to clients (the reverse of normal client -> server flow).

WARNING: These APIs are experimental and may change without notice.

Use cases: - Server sends task-augmented sampling/elicitation request to client - Client creates a local task, spawns background work, returns CreateTaskResult - Server polls client's task status via tasks/get, tasks/result, etc.

GetTaskHandlerFnT

Bases: Protocol

Handler for tasks/get requests from server.

WARNING: This is experimental and may change without notice.

Source code in src/mcp/client/experimental/task_handlers.py 
29
30
31
32
33
34
35
36
37
38
39
class GetTaskHandlerFnT(Protocol):
    """Handler for tasks/get requests from server.

    WARNING: This is experimental and may change without notice.
    """

    async def __call__(
        self,
        context: RequestContext[ClientSession],
        params: types.GetTaskRequestParams,
    ) -> types.GetTaskResult | types.ErrorData: ...  # pragma: no branch

GetTaskResultHandlerFnT

Bases: Protocol

Handler for tasks/result requests from server.

WARNING: This is experimental and may change without notice.

Source code in src/mcp/client/experimental/task_handlers.py 
42
43
44
45
46
47
48
49
50
51
52
class GetTaskResultHandlerFnT(Protocol):
    """Handler for tasks/result requests from server.

    WARNING: This is experimental and may change without notice.
    """

    async def __call__(
        self,
        context: RequestContext[ClientSession],
        params: types.GetTaskPayloadRequestParams,
    ) -> types.GetTaskPayloadResult | types.ErrorData: ...  # pragma: no branch

ListTasksHandlerFnT

Bases: Protocol

Handler for tasks/list requests from server.

WARNING: This is experimental and may change without notice.

Source code in src/mcp/client/experimental/task_handlers.py 
55
56
57
58
59
60
61
62
63
64
65
class ListTasksHandlerFnT(Protocol):
    """Handler for tasks/list requests from server.

    WARNING: This is experimental and may change without notice.
    """

    async def __call__(
        self,
        context: RequestContext[ClientSession],
        params: types.PaginatedRequestParams | None,
    ) -> types.ListTasksResult | types.ErrorData: ...  # pragma: no branch

CancelTaskHandlerFnT

Bases: Protocol

Handler for tasks/cancel requests from server.

WARNING: This is experimental and may change without notice.

Source code in src/mcp/client/experimental/task_handlers.py 
68
69
70
71
72
73
74
75
76
77
78
class CancelTaskHandlerFnT(Protocol):
    """Handler for tasks/cancel requests from server.

    WARNING: This is experimental and may change without notice.
    """

    async def __call__(
        self,
        context: RequestContext[ClientSession],
        params: types.CancelTaskRequestParams,
    ) -> types.CancelTaskResult | types.ErrorData: ...  # pragma: no branch

TaskAugmentedSamplingFnT

Bases: Protocol

Handler for task-augmented sampling/createMessage requests from server.

When server sends a CreateMessageRequest with task field, this callback is invoked. The callback should create a task, spawn background work, and return CreateTaskResult immediately.

WARNING: This is experimental and may change without notice.

Source code in src/mcp/client/experimental/task_handlers.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class TaskAugmentedSamplingFnT(Protocol):
    """Handler for task-augmented sampling/createMessage requests from server.

    When server sends a CreateMessageRequest with task field, this callback
    is invoked. The callback should create a task, spawn background work,
    and return CreateTaskResult immediately.

    WARNING: This is experimental and may change without notice.
    """

    async def __call__(
        self,
        context: RequestContext[ClientSession],
        params: types.CreateMessageRequestParams,
        task_metadata: types.TaskMetadata,
    ) -> types.CreateTaskResult | types.ErrorData: ...  # pragma: no branch

TaskAugmentedElicitationFnT

Bases: Protocol

Handler for task-augmented elicitation/create requests from server.

When server sends an ElicitRequest with task field, this callback is invoked. The callback should create a task, spawn background work, and return CreateTaskResult immediately.

WARNING: This is experimental and may change without notice.

Source code in src/mcp/client/experimental/task_handlers.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class TaskAugmentedElicitationFnT(Protocol):
    """Handler for task-augmented elicitation/create requests from server.

    When server sends an ElicitRequest with task field, this callback
    is invoked. The callback should create a task, spawn background work,
    and return CreateTaskResult immediately.

    WARNING: This is experimental and may change without notice.
    """

    async def __call__(
        self,
        context: RequestContext[ClientSession],
        params: types.ElicitRequestParams,
        task_metadata: types.TaskMetadata,
    ) -> types.CreateTaskResult | types.ErrorData: ...  # pragma: no branch

ExperimentalTaskHandlers dataclass

Container for experimental task handlers.

Groups all task-related handlers that handle server -> client requests. This includes both pure task requests (get, list, cancel, result) and task-augmented request handlers (sampling, elicitation with task field).

WARNING: These APIs are experimental and may change without notice.

Example 
handlers = ExperimentalTaskHandlers(
    get_task=my_get_task_handler,
    list_tasks=my_list_tasks_handler,
)
session = ClientSession(..., experimental_task_handlers=handlers)
Source code in src/mcp/client/experimental/task_handlers.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
@dataclass
class ExperimentalTaskHandlers:
    """Container for experimental task handlers.

    Groups all task-related handlers that handle server -> client requests.
    This includes both pure task requests (get, list, cancel, result) and
    task-augmented request handlers (sampling, elicitation with task field).

    WARNING: These APIs are experimental and may change without notice.

    Example:
        ```python
        handlers = ExperimentalTaskHandlers(
            get_task=my_get_task_handler,
            list_tasks=my_list_tasks_handler,
        )
        session = ClientSession(..., experimental_task_handlers=handlers)
        ```
    """

    # Pure task request handlers
    get_task: GetTaskHandlerFnT = field(default=default_get_task_handler)
    get_task_result: GetTaskResultHandlerFnT = field(default=default_get_task_result_handler)
    list_tasks: ListTasksHandlerFnT = field(default=default_list_tasks_handler)
    cancel_task: CancelTaskHandlerFnT = field(default=default_cancel_task_handler)

    # Task-augmented request handlers
    augmented_sampling: TaskAugmentedSamplingFnT = field(default=default_task_augmented_sampling)
    augmented_elicitation: TaskAugmentedElicitationFnT = field(default=default_task_augmented_elicitation)

    def build_capability(self) -> types.ClientTasksCapability | None:
        """Build ClientTasksCapability from the configured handlers.

        Returns a capability object that reflects which handlers are configured
        (i.e., not using the default "not supported" handlers).

        Returns:
            ClientTasksCapability if any handlers are provided, None otherwise
        """
        has_list = self.list_tasks is not default_list_tasks_handler
        has_cancel = self.cancel_task is not default_cancel_task_handler
        has_sampling = self.augmented_sampling is not default_task_augmented_sampling
        has_elicitation = self.augmented_elicitation is not default_task_augmented_elicitation

        # If no handlers are provided, return None
        if not any([has_list, has_cancel, has_sampling, has_elicitation]):
            return None

        # Build requests capability if any request handlers are provided
        requests_capability: types.ClientTasksRequestsCapability | None = None
        if has_sampling or has_elicitation:
            requests_capability = types.ClientTasksRequestsCapability(
                sampling=types.TasksSamplingCapability(create_message=types.TasksCreateMessageCapability())
                if has_sampling
                else None,
                elicitation=types.TasksElicitationCapability(create=types.TasksCreateElicitationCapability())
                if has_elicitation
                else None,
            )

        return types.ClientTasksCapability(
            list=types.TasksListCapability() if has_list else None,
            cancel=types.TasksCancelCapability() if has_cancel else None,
            requests=requests_capability,
        )

    @staticmethod
    def handles_request(request: types.ServerRequest) -> bool:
        """Check if this handler handles the given request type."""
        return isinstance(
            request,
            types.GetTaskRequest | types.GetTaskPayloadRequest | types.ListTasksRequest | types.CancelTaskRequest,
        )

    async def handle_request(
        self,
        ctx: RequestContext[ClientSession],
        responder: RequestResponder[types.ServerRequest, types.ClientResult],
    ) -> None:
        """Handle a task-related request from the server.

        Call handles_request() first to check if this handler can handle the request.
        """
        client_response_type: TypeAdapter[types.ClientResult | types.ErrorData] = TypeAdapter(
            types.ClientResult | types.ErrorData
        )

        match responder.request:
            case types.GetTaskRequest(params=params):
                response = await self.get_task(ctx, params)
                client_response = client_response_type.validate_python(response)
                await responder.respond(client_response)

            case types.GetTaskPayloadRequest(params=params):
                response = await self.get_task_result(ctx, params)
                client_response = client_response_type.validate_python(response)
                await responder.respond(client_response)

            case types.ListTasksRequest(params=params):
                response = await self.list_tasks(ctx, params)
                client_response = client_response_type.validate_python(response)
                await responder.respond(client_response)

            case types.CancelTaskRequest(params=params):
                response = await self.cancel_task(ctx, params)
                client_response = client_response_type.validate_python(response)
                await responder.respond(client_response)

            case _:  # pragma: no cover
                raise ValueError(f"Unhandled request type: {type(responder.request)}")

build_capability 

build_capability() -> ClientTasksCapability | None

Build ClientTasksCapability from the configured handlers.

Returns a capability object that reflects which handlers are configured (i.e., not using the default "not supported" handlers).

Returns:

Type Description
ClientTasksCapability | None

ClientTasksCapability if any handlers are provided, None otherwise
Source code in src/mcp/client/experimental/task_handlers.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def build_capability(self) -> types.ClientTasksCapability | None:
    """Build ClientTasksCapability from the configured handlers.

    Returns a capability object that reflects which handlers are configured
    (i.e., not using the default "not supported" handlers).

    Returns:
        ClientTasksCapability if any handlers are provided, None otherwise
    """
    has_list = self.list_tasks is not default_list_tasks_handler
    has_cancel = self.cancel_task is not default_cancel_task_handler
    has_sampling = self.augmented_sampling is not default_task_augmented_sampling
    has_elicitation = self.augmented_elicitation is not default_task_augmented_elicitation

    # If no handlers are provided, return None
    if not any([has_list, has_cancel, has_sampling, has_elicitation]):
        return None

    # Build requests capability if any request handlers are provided
    requests_capability: types.ClientTasksRequestsCapability | None = None
    if has_sampling or has_elicitation:
        requests_capability = types.ClientTasksRequestsCapability(
            sampling=types.TasksSamplingCapability(create_message=types.TasksCreateMessageCapability())
            if has_sampling
            else None,
            elicitation=types.TasksElicitationCapability(create=types.TasksCreateElicitationCapability())
            if has_elicitation
            else None,
        )

    return types.ClientTasksCapability(
        list=types.TasksListCapability() if has_list else None,
        cancel=types.TasksCancelCapability() if has_cancel else None,
        requests=requests_capability,
    )

handles_request staticmethod 

handles_request(request: ServerRequest) -> bool

Check if this handler handles the given request type.

Source code in src/mcp/client/experimental/task_handlers.py 
245
246
247
248
249
250
251
@staticmethod
def handles_request(request: types.ServerRequest) -> bool:
    """Check if this handler handles the given request type."""
    return isinstance(
        request,
        types.GetTaskRequest | types.GetTaskPayloadRequest | types.ListTasksRequest | types.CancelTaskRequest,
    )

handle_request async 

handle_request(
    ctx: RequestContext[ClientSession],
    responder: RequestResponder[
        ServerRequest, ClientResult
    ],
) -> None

Handle a task-related request from the server.

Call handles_request() first to check if this handler can handle the request.

Source code in src/mcp/client/experimental/task_handlers.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
async def handle_request(
    self,
    ctx: RequestContext[ClientSession],
    responder: RequestResponder[types.ServerRequest, types.ClientResult],
) -> None:
    """Handle a task-related request from the server.

    Call handles_request() first to check if this handler can handle the request.
    """
    client_response_type: TypeAdapter[types.ClientResult | types.ErrorData] = TypeAdapter(
        types.ClientResult | types.ErrorData
    )

    match responder.request:
        case types.GetTaskRequest(params=params):
            response = await self.get_task(ctx, params)
            client_response = client_response_type.validate_python(response)
            await responder.respond(client_response)

        case types.GetTaskPayloadRequest(params=params):
            response = await self.get_task_result(ctx, params)
            client_response = client_response_type.validate_python(response)
            await responder.respond(client_response)

        case types.ListTasksRequest(params=params):
            response = await self.list_tasks(ctx, params)
            client_response = client_response_type.validate_python(response)
            await responder.respond(client_response)

        case types.CancelTaskRequest(params=params):
            response = await self.cancel_task(ctx, params)
            client_response = client_response_type.validate_python(response)
            await responder.respond(client_response)

        case _:  # pragma: no cover
            raise ValueError(f"Unhandled request type: {type(responder.request)}")