notifications and client side

Author: maxisbey

examples/servers/simple-task/mcp_simple_task/server.py

@@ -8,6 +8,7 @@
 import anyio
 import click
 import mcp.types as types
+import uvicorn
 from anyio.abc import TaskGroup
 from mcp.server.lowlevel import Server
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
@@ -107,8 +108,6 @@ async def handle_get_task_result(request: types.GetTaskPayloadRequest) -> types.
 @click.command()
 @click.option("--port", default=8000, help="Port to listen on")
 def main(port: int) -> int:
-    import uvicorn
-
     session_manager = StreamableHTTPSessionManager(app=server)
 
     @asynccontextmanager

src/mcp/client/session.py

@@ -49,6 +49,95 @@ async def __call__(
     ) -> None: ...  # pragma: no branch
 
 
+# Experimental: Task handler protocols for server -> client requests
+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", Any],
+        params: types.GetTaskRequestParams,
+    ) -> types.GetTaskResult | types.ErrorData: ...  # pragma: no branch
+
+
+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", Any],
+        params: types.GetTaskPayloadRequestParams,
+    ) -> types.GetTaskPayloadResult | types.ErrorData: ...  # pragma: no branch
+
+
+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", Any],
+        params: types.PaginatedRequestParams | None,
+    ) -> types.ListTasksResult | types.ErrorData: ...  # pragma: no branch
+
+
+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", Any],
+        params: types.CancelTaskRequestParams,
+    ) -> types.CancelTaskResult | types.ErrorData: ...  # pragma: no branch
+
+
+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", Any],
+        params: types.CreateMessageRequestParams,
+        task_metadata: types.TaskMetadata,
+    ) -> types.CreateTaskResult | types.ErrorData: ...  # pragma: no branch
+
+
+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", Any],
+        params: types.ElicitRequestParams,
+        task_metadata: types.TaskMetadata,
+    ) -> types.CreateTaskResult | types.ErrorData: ...  # pragma: no branch
+
+
 class MessageHandlerFnT(Protocol):
     async def __call__(
         self,
@@ -97,6 +186,69 @@ async def _default_logging_callback(
     pass
 
 
+# Default handlers for experimental task requests (return "not supported" errors)
+async def _default_get_task_handler(
+    context: RequestContext["ClientSession", Any],
+    params: types.GetTaskRequestParams,
+) -> types.GetTaskResult | types.ErrorData:
+    return types.ErrorData(
+        code=types.METHOD_NOT_FOUND,
+        message="tasks/get not supported",
+    )
+
+
+async def _default_get_task_result_handler(
+    context: RequestContext["ClientSession", Any],
+    params: types.GetTaskPayloadRequestParams,
+) -> types.GetTaskPayloadResult | types.ErrorData:
+    return types.ErrorData(
+        code=types.METHOD_NOT_FOUND,
+        message="tasks/result not supported",
+    )
+
+
+async def _default_list_tasks_handler(
+    context: RequestContext["ClientSession", Any],
+    params: types.PaginatedRequestParams | None,
+) -> types.ListTasksResult | types.ErrorData:
+    return types.ErrorData(
+        code=types.METHOD_NOT_FOUND,
+        message="tasks/list not supported",
+    )
+
+
+async def _default_cancel_task_handler(
+    context: RequestContext["ClientSession", Any],
+    params: types.CancelTaskRequestParams,
+) -> types.CancelTaskResult | types.ErrorData:
+    return types.ErrorData(
+        code=types.METHOD_NOT_FOUND,
+        message="tasks/cancel not supported",
+    )
+
+
+async def _default_task_augmented_sampling_callback(
+    context: RequestContext["ClientSession", Any],
+    params: types.CreateMessageRequestParams,
+    task_metadata: types.TaskMetadata,
+) -> types.CreateTaskResult | types.ErrorData:
+    return types.ErrorData(
+        code=types.INVALID_REQUEST,
+        message="Task-augmented sampling not supported",
+    )
+
+
+async def _default_task_augmented_elicitation_callback(
+    context: RequestContext["ClientSession", Any],
+    params: types.ElicitRequestParams,
+    task_metadata: types.TaskMetadata,
+) -> types.CreateTaskResult | types.ErrorData:
+    return types.ErrorData(
+        code=types.INVALID_REQUEST,
+        message="Task-augmented elicitation not supported",
+    )
+
+
 ClientResponse: TypeAdapter[types.ClientResult | types.ErrorData] = TypeAdapter(types.ClientResult | types.ErrorData)
 
 
@@ -120,6 +272,14 @@ def __init__(
         logging_callback: LoggingFnT | None = None,
         message_handler: MessageHandlerFnT | None = None,
         client_info: types.Implementation | None = None,
+        tasks_capability: types.ClientTasksCapability | None = None,
+        # Experimental: Task handlers for server -> client requests
+        get_task_handler: GetTaskHandlerFnT | None = None,
+        get_task_result_handler: GetTaskResultHandlerFnT | None = None,
+        list_tasks_handler: ListTasksHandlerFnT | None = None,
+        cancel_task_handler: CancelTaskHandlerFnT | None = None,
+        task_augmented_sampling_callback: TaskAugmentedSamplingFnT | None = None,
+        task_augmented_elicitation_callback: TaskAugmentedElicitationFnT | None = None,
     ) -> None:
         super().__init__(
             read_stream,
@@ -134,9 +294,21 @@ def __init__(
         self._list_roots_callback = list_roots_callback or _default_list_roots_callback
         self._logging_callback = logging_callback or _default_logging_callback
         self._message_handler = message_handler or _default_message_handler
+        self._tasks_capability = tasks_capability
         self._tool_output_schemas: dict[str, dict[str, Any] | None] = {}
         self._server_capabilities: types.ServerCapabilities | None = None
         self._experimental: ExperimentalClientFeatures | None = None
+        # Experimental: Task handlers
+        self._get_task_handler = get_task_handler or _default_get_task_handler
+        self._get_task_result_handler = get_task_result_handler or _default_get_task_result_handler
+        self._list_tasks_handler = list_tasks_handler or _default_list_tasks_handler
+        self._cancel_task_handler = cancel_task_handler or _default_cancel_task_handler
+        self._task_augmented_sampling_callback = (
+            task_augmented_sampling_callback or _default_task_augmented_sampling_callback
+        )
+        self._task_augmented_elicitation_callback = (
+            task_augmented_elicitation_callback or _default_task_augmented_elicitation_callback
+        )
 
     async def initialize(self) -> types.InitializeResult:
         sampling = types.SamplingCapability() if self._sampling_callback is not _default_sampling_callback else None
@@ -162,6 +334,7 @@ async def initialize(self) -> types.InitializeResult:
                             elicitation=elicitation,
                             experimental=None,
                             roots=roots,
+                            tasks=self._tasks_capability,
                         ),
                         clientInfo=self._client_info,
                     ),
@@ -187,7 +360,7 @@ def get_server_capabilities(self) -> types.ServerCapabilities | None:
         return self._server_capabilities
 
     @property
-    def experimental(self) -> "ExperimentalClientFeatures":
+    def experimental(self) -> ExperimentalClientFeatures:
         """Experimental APIs for tasks and other features.
 
         WARNING: These APIs are experimental and may change without notice.
@@ -534,13 +707,21 @@ async def _received_request(self, responder: RequestResponder[types.ServerReques
         match responder.request.root:
             case types.CreateMessageRequest(params=params):
                 with responder:
-                    response = await self._sampling_callback(ctx, params)
+                    # Check if this is a task-augmented request
+                    if params.task is not None:
+                        response = await self._task_augmented_sampling_callback(ctx, params, params.task)
+                    else:
+                        response = await self._sampling_callback(ctx, params)
                     client_response = ClientResponse.validate_python(response)
                     await responder.respond(client_response)
 
             case types.ElicitRequest(params=params):
                 with responder:
-                    response = await self._elicitation_callback(ctx, params)
+                    # Check if this is a task-augmented request
+                    if params.task is not None:
+                        response = await self._task_augmented_elicitation_callback(ctx, params, params.task)
+                    else:
+                        response = await self._elicitation_callback(ctx, params)
                     client_response = ClientResponse.validate_python(response)
                     await responder.respond(client_response)
 
@@ -553,7 +734,33 @@ async def _received_request(self, responder: RequestResponder[types.ServerReques
             case types.PingRequest():  # pragma: no cover
                 with responder:
                     return await responder.respond(types.ClientResult(root=types.EmptyResult()))
-            case _:
+
+            # Experimental: Task management requests from server
+            case types.GetTaskRequest(params=params):
+                with responder:
+                    response = await self._get_task_handler(ctx, params)
+                    client_response = ClientResponse.validate_python(response)
+                    await responder.respond(client_response)
+
+            case types.GetTaskPayloadRequest(params=params):
+                with responder:
+                    response = await self._get_task_result_handler(ctx, params)
+                    client_response = ClientResponse.validate_python(response)
+                    await responder.respond(client_response)
+
+            case types.ListTasksRequest(params=params):
+                with responder:
+                    response = await self._list_tasks_handler(ctx, params)
+                    client_response = ClientResponse.validate_python(response)
+                    await responder.respond(client_response)
+
+            case types.CancelTaskRequest(params=params):
+                with responder:
+                    response = await self._cancel_task_handler(ctx, params)
+                    client_response = ClientResponse.validate_python(response)
+                    await responder.respond(client_response)
+
+            case _:  # pragma: no cover
                 raise NotImplementedError()
 
     async def _handle_incoming(

src/mcp/server/lowlevel/server.py

@@ -67,12 +67,14 @@ async def main():
 
 from __future__ import annotations as _annotations
 
+import base64
 import contextvars
 import json
 import logging
 import warnings
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
+from importlib.metadata import version as pkg_version
 from typing import Any, Generic, TypeAlias, cast
 
 import anyio
@@ -165,19 +167,17 @@ def create_initialization_options(
     ) -> InitializationOptions:
         """Create initialization options from this server instance."""
 
-        def pkg_version(package: str) -> str:
+        def get_package_version(package: str) -> str:
             try:
-                from importlib.metadata import version
-
-                return version(package)
+                return pkg_version(package)
             except Exception:  # pragma: no cover
                 pass
 
             return "unknown"  # pragma: no cover
 
         return InitializationOptions(
             server_name=self.name,
-            server_version=self.version if self.version else pkg_version("mcp"),
+            server_version=self.version if self.version else get_package_version("mcp"),
             capabilities=self.get_capabilities(
                 notification_options or NotificationOptions(),
                 experimental_capabilities or {},
@@ -344,8 +344,6 @@ def create_content(data: str | bytes, mime_type: str | None):
                                 mimeType=mime_type or "text/plain",
                             )
                         case bytes() as data:  # pragma: no cover
-                            import base64
-
                             return types.BlobResourceContents(
                                 uri=req.params.uri,
                                 blob=base64.b64encode(data).decode(),

src/mcp/shared/experimental/tasks/__init__.py

@@ -5,10 +5,13 @@
 - TaskStore: Abstract interface for task state storage
 - TaskContext: Context object for task work to interact with state/notifications
 - InMemoryTaskStore: Reference implementation for testing/development
+- TaskMessageQueue: FIFO queue for task messages delivered via tasks/result
+- InMemoryTaskMessageQueue: Reference implementation for message queue
 - Helper functions: run_task, is_terminal, create_task_state, generate_task_id
 
 Architecture:
 - TaskStore is pure storage - it doesn't know about execution
+- TaskMessageQueue stores messages to be delivered via tasks/result
 - TaskContext wraps store + session, providing a clean API for task work
 - run_task is optional convenience for spawning in-process tasks
 
@@ -24,15 +27,31 @@
     task_execution,
 )
 from mcp.shared.experimental.tasks.in_memory_task_store import InMemoryTaskStore
+from mcp.shared.experimental.tasks.message_queue import (
+    InMemoryTaskMessageQueue,
+    QueuedMessage,
+    TaskMessageQueue,
+)
+from mcp.shared.experimental.tasks.result_handler import (
+    TaskResultHandler,
+    create_task_result_handler,
+)
 from mcp.shared.experimental.tasks.store import TaskStore
+from mcp.shared.experimental.tasks.task_session import TaskSession
 
 __all__ = [
     "TaskStore",
     "TaskContext",
+    "TaskSession",
+    "TaskResultHandler",
     "InMemoryTaskStore",
+    "TaskMessageQueue",
+    "InMemoryTaskMessageQueue",
+    "QueuedMessage",
     "run_task",
     "task_execution",
     "is_terminal",
     "create_task_state",
     "generate_task_id",
+    "create_task_result_handler",
 ]