Make session_id optional in TaskStore; reject stateless mode instead

Previously, session_id was required (str) throughout the task store
interface, which forced sessionless transports like stdio and memory
to fabricate UUIDs before running tasks. This was awkward because
those transports are single-client by architecture and have no session
concept.

The policy is now enforced at the handler layer instead of the store
layer: default task handlers reject requests when the server is in
stateless mode (where tasks cannot survive across requests), and pass
session_id through as-is otherwise. A None session_id simply means no
session-scoped isolation, which is correct for single-client transports.

Isolation in InMemoryTaskStore uses strict equality: None only matches
None, so tasks created by a sessionless transport are not visible to
session-scoped transports and vice versa, preventing cross-transport
leaks when a process serves multiple transports from one store.

Changes:
- TaskStore, InMemoryTaskStore, TaskContext, helpers: session_id is
  now str | None
- ServerSession: expose stateless property
- Default task handlers: reject stateless mode, pass session_id as-is
- run_task() / ServerTaskContext: accept None session_id, reject
  stateless mode
- Memory transport: revert fabricated UUID session_id
- New tests for None-session isolation (strict equality, no
  cross-transport leaks)

:house: Remote-Dev: homespace

Author: localden

src/mcp/client/_memory.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import uuid
 from collections.abc import AsyncIterator
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from types import TracebackType
@@ -51,14 +50,12 @@ async def _connect(self) -> AsyncIterator[TransportStreams]:
 
             async with anyio.create_task_group() as tg:
                 # Start server in background
-                memory_session_id = uuid.uuid4().hex
                 tg.start_soon(
                     lambda: actual_server.run(
                         server_read,
                         server_write,
                         actual_server.create_initialization_options(),
                         raise_exceptions=self._raise_exceptions,
-                        session_id=memory_session_id,
                     )
                 )
 

src/mcp/server/experimental/request_context.py

@@ -189,9 +189,11 @@ async def work(task: ServerTaskContext) -> CallToolResult:
         # Access task_group via TaskSupport - raises if not in run() context
         task_group = support.task_group
 
+        if self._session.stateless:
+            raise RuntimeError(
+                "run_task() does not support stateless mode. Tasks require a persistent session for result retrieval."
+            )
         session_id = self._session.session_id
-        if session_id is None:
-            raise RuntimeError("Session ID is required for task operations but session has no ID.")
         task = await support.store.create_task(self.task_metadata, task_id, session_id=session_id)
 
         task_ctx = ServerTaskContext(

src/mcp/server/experimental/task_context.py

@@ -90,11 +90,8 @@ def __init__(
             queue: The message queue for elicitation/sampling
             handler: The result handler for response routing (required for elicit/create_message)
         """
-        session_id = session.session_id
-        if session_id is None:
-            raise RuntimeError("Session ID is required for task operations but session has no ID.")
-        self._session_id = session_id
-        self._ctx = TaskContext(task=task, store=store, session_id=session_id)
+        self._session_id = session.session_id
+        self._ctx = TaskContext(task=task, store=store, session_id=self._session_id)
         self._session = session
         self._queue = queue
         self._handler = handler

src/mcp/server/experimental/task_result_handler.py

@@ -80,7 +80,7 @@ async def handle(
         request: GetTaskPayloadRequest,
         session: ServerSession,
         request_id: RequestId,
-        session_id: str,
+        session_id: str | None,
     ) -> GetTaskPayloadResult:
         """Handle a tasks/result request.
 
@@ -95,7 +95,8 @@ async def handle(
             request: The GetTaskPayloadRequest
             session: The server session for sending messages
             request_id: The request ID for relatedRequestId routing
-            session_id: Session identifier for access control.
+            session_id: Session identifier for access control. Must exactly
+                match the session_id the task was created with (including None).
 
         Returns:
             GetTaskPayloadResult with the task's final payload

src/mcp/server/lowlevel/experimental.py

@@ -153,23 +153,24 @@ def enable_tasks(
         if on_cancel_task is not None:
             self._add_request_handler("tasks/cancel", on_cancel_task)
 
-        def _require_session_id(ctx: ServerRequestContext[LifespanResultT]) -> str:
-            session_id = ctx.session.session_id
-            if session_id is None:
+        def _check_stateless(ctx: ServerRequestContext[LifespanResultT]) -> None:
+            if ctx.session.stateless:
                 raise MCPError(
                     code=INVALID_PARAMS,
-                    message="Session ID is required for task operations.",
+                    message=(
+                        "Default task handlers do not support stateless mode. "
+                        "Provide custom task handlers if you need stateless task support."
+                    ),
                 )
-            return session_id
 
         # Fill in defaults for any not provided
         if not self._has_handler("tasks/get"):
 
             async def _default_get_task(
                 ctx: ServerRequestContext[LifespanResultT], params: GetTaskRequestParams
             ) -> GetTaskResult:
-                session_id = _require_session_id(ctx)
-                task = await task_support.store.get_task(params.task_id, session_id=session_id)
+                _check_stateless(ctx)
+                task = await task_support.store.get_task(params.task_id, session_id=ctx.session.session_id)
                 if task is None:
                     raise MCPError(code=INVALID_PARAMS, message=f"Task not found: {params.task_id}")
                 return GetTaskResult(
@@ -190,9 +191,11 @@ async def _default_get_task_result(
                 ctx: ServerRequestContext[LifespanResultT], params: GetTaskPayloadRequestParams
             ) -> GetTaskPayloadResult:
                 assert ctx.request_id is not None
-                session_id = _require_session_id(ctx)
+                _check_stateless(ctx)
                 req = GetTaskPayloadRequest(params=params)
-                result = await task_support.handler.handle(req, ctx.session, ctx.request_id, session_id=session_id)
+                result = await task_support.handler.handle(
+                    req, ctx.session, ctx.request_id, session_id=ctx.session.session_id
+                )
                 return result
 
             self._add_request_handler("tasks/result", _default_get_task_result)
@@ -202,9 +205,9 @@ async def _default_get_task_result(
             async def _default_list_tasks(
                 ctx: ServerRequestContext[LifespanResultT], params: PaginatedRequestParams | None
             ) -> ListTasksResult:
+                _check_stateless(ctx)
                 cursor = params.cursor if params else None
-                session_id = _require_session_id(ctx)
-                tasks, next_cursor = await task_support.store.list_tasks(cursor, session_id=session_id)
+                tasks, next_cursor = await task_support.store.list_tasks(cursor, session_id=ctx.session.session_id)
                 return ListTasksResult(tasks=tasks, next_cursor=next_cursor)
 
             self._add_request_handler("tasks/list", _default_list_tasks)
@@ -214,8 +217,8 @@ async def _default_list_tasks(
             async def _default_cancel_task(
                 ctx: ServerRequestContext[LifespanResultT], params: CancelTaskRequestParams
             ) -> CancelTaskResult:
-                session_id = _require_session_id(ctx)
-                result = await cancel_task(task_support.store, params.task_id, session_id=session_id)
+                _check_stateless(ctx)
+                result = await cancel_task(task_support.store, params.task_id, session_id=ctx.session.session_id)
                 return result
 
             self._add_request_handler("tasks/cancel", _default_cancel_task)

src/mcp/server/session.py

@@ -111,6 +111,11 @@ def _receive_notification_adapter(self) -> TypeAdapter[types.ClientNotification]
     def client_params(self) -> types.InitializeRequestParams | None:
         return self._client_params
 
+    @property
+    def stateless(self) -> bool:
+        """Whether this session is in stateless mode (no persistent server-side state)."""
+        return self._stateless
+
     @property
     def experimental(self) -> ExperimentalServerSessionFeatures:
         """Experimental APIs for server→client task operations.

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

@@ -21,7 +21,7 @@ class TaskContext:
     use ServerTaskContext from mcp.server.experimental.
 
     Example (distributed worker):
-        async def worker_job(task_id: str, session_id: str):
+        async def worker_job(task_id: str, session_id: str | None):
             store = RedisTaskStore(redis_url)
             task = await store.get_task(task_id, session_id=session_id)
             ctx = TaskContext(task=task, store=store, session_id=session_id)
@@ -31,7 +31,7 @@ async def worker_job(task_id: str, session_id: str):
             await ctx.complete(result)
     """
 
-    def __init__(self, task: Task, store: TaskStore, *, session_id: str):
+    def __init__(self, task: Task, store: TaskStore, *, session_id: str | None):
         self._task = task
         self._store = store
         self._session_id = session_id

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

@@ -51,7 +51,7 @@ async def cancel_task(
     store: TaskStore,
     task_id: str,
     *,
-    session_id: str,
+    session_id: str | None,
 ) -> CancelTaskResult:
     """Cancel a task with spec-compliant validation.
 
@@ -64,7 +64,8 @@ async def cancel_task(
     Args:
         store: The task store
         task_id: The task identifier to cancel
-        session_id: Session identifier for access control.
+        session_id: Session identifier for access control. Must exactly match
+            the session_id the task was created with (including None).
 
     Returns:
         CancelTaskResult with the cancelled task state
@@ -128,7 +129,7 @@ async def task_execution(
     task_id: str,
     store: TaskStore,
     *,
-    session_id: str,
+    session_id: str | None,
 ) -> AsyncIterator[TaskContext]:
     """Context manager for safe task execution (pure, no server dependencies).
 
@@ -141,7 +142,8 @@ async def task_execution(
     Args:
         task_id: The task identifier to execute
         store: The task store (must be accessible by the worker)
-        session_id: Session identifier for access control.
+        session_id: Session identifier for access control. Must exactly match
+            the session_id the task was created with (including None).
 
     Yields:
         TaskContext for updating status and completing/failing the task
@@ -150,7 +152,7 @@ async def task_execution(
         ValueError: If the task is not found in the store
 
     Example (distributed worker):
-        async def worker_process(task_id: str, session_id: str):
+        async def worker_process(task_id: str, session_id: str | None):
             store = RedisTaskStore(redis_url)
             async with task_execution(task_id, store, session_id=session_id) as ctx:
                 await ctx.update_status("Working...")

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

@@ -22,7 +22,7 @@ class StoredTask:
     """Internal storage representation of a task."""
 
     task: Task
-    session_id: str
+    session_id: str | None
     result: Result | None = None
     # Time when this task should be removed (None = never)
     expires_at: datetime | None = field(default=None)
@@ -68,10 +68,13 @@ def _cleanup_expired(self) -> None:
         for task_id in expired_ids:
             del self._tasks[task_id]
 
-    def _get_stored_task(self, task_id: str, *, session_id: str) -> StoredTask | None:
+    def _get_stored_task(self, task_id: str, *, session_id: str | None) -> StoredTask | None:
         """Retrieve a stored task, enforcing session ownership.
 
         Returns None if the task does not exist or belongs to a different session.
+        Isolation uses strict equality: None only matches None, so tasks created
+        by sessionless transports (stdio) are not visible to session-scoped
+        transports (HTTP) and vice versa.
         """
         stored = self._tasks.get(task_id)
         if stored is None:
@@ -85,7 +88,7 @@ async def create_task(
         metadata: TaskMetadata,
         task_id: str | None = None,
         *,
-        session_id: str,
+        session_id: str | None,
     ) -> Task:
         """Create a new task with the given metadata."""
         # Cleanup expired tasks on access
@@ -106,7 +109,7 @@ async def create_task(
         # Return a copy to prevent external modification
         return Task(**task.model_dump())
 
-    async def get_task(self, task_id: str, *, session_id: str) -> Task | None:
+    async def get_task(self, task_id: str, *, session_id: str | None) -> Task | None:
         """Get a task by ID."""
         # Cleanup expired tasks on access
         self._cleanup_expired()
@@ -124,7 +127,7 @@ async def update_task(
         status: TaskStatus | None = None,
         status_message: str | None = None,
         *,
-        session_id: str,
+        session_id: str | None,
     ) -> Task:
         """Update a task's status and/or message."""
         stored = self._get_stored_task(task_id, session_id=session_id)
@@ -156,15 +159,15 @@ async def update_task(
 
         return Task(**stored.task.model_dump())
 
-    async def store_result(self, task_id: str, result: Result, *, session_id: str) -> None:
+    async def store_result(self, task_id: str, result: Result, *, session_id: str | None) -> None:
         """Store the result for a task."""
         stored = self._get_stored_task(task_id, session_id=session_id)
         if stored is None:
             raise ValueError(f"Task with ID {task_id} not found")
 
         stored.result = result
 
-    async def get_result(self, task_id: str, *, session_id: str) -> Result | None:
+    async def get_result(self, task_id: str, *, session_id: str | None) -> Result | None:
         """Get the stored result for a task."""
         stored = self._get_stored_task(task_id, session_id=session_id)
         if stored is None:
@@ -176,13 +179,14 @@ async def list_tasks(
         self,
         cursor: str | None = None,
         *,
-        session_id: str,
+        session_id: str | None,
     ) -> tuple[list[Task], str | None]:
         """List tasks with pagination."""
         # Cleanup expired tasks on access
         self._cleanup_expired()
 
-        # Filter tasks by session ownership before pagination
+        # Filter tasks by session ownership before pagination.
+        # Strict equality: None only matches None (sessionless transports).
         filtered_task_ids = [task_id for task_id, stored in self._tasks.items() if stored.session_id == session_id]
 
         start_index = 0
@@ -203,7 +207,7 @@ async def list_tasks(
 
         return tasks, next_cursor
 
-    async def delete_task(self, task_id: str, *, session_id: str) -> bool:
+    async def delete_task(self, task_id: str, *, session_id: str | None) -> bool:
         """Delete a task."""
         stored = self._get_stored_task(task_id, session_id=session_id)
         if stored is None: