MCP long-running task support in Python

python/packages/core/AGENTS.md

@@ -76,6 +76,16 @@ agent_framework/
 - **`SkillScriptRunner`** - Protocol for file-based script execution. Any callable matching `(skill, script, args) -> Any` satisfies it. Code-defined scripts do not use a runner.
 - **`SkillsProvider`** - Context provider (extends `ContextProvider`) that discovers file-based skills from `SKILL.md` files and/or accepts code-defined `Skill` instances. Follows progressive disclosure: advertise → load → read resources / run scripts.
 
+### Model Context Protocol (`_mcp.py`)
+
+- **`MCPTool`** - Base wrapper that owns the MCP `ClientSession` and exposes the remote server's tools as `FunctionTool`s.
+- **`MCPStdioTool`** / **`MCPStreamableHTTPTool`** / **`MCPWebsocketTool`** - Transport-specific subclasses.
+- **`MCPTaskOptions`** (experimental, `MCP_LONG_RUNNING_TASKS` feature) - Per-tool-instance options controlling the SEP-2663 long-running task lifecycle. When the server advertises a tool with `execution.taskSupport == "required"`, `MCPTool.call_tool` transparently routes through `call_tool_as_task`, which sends an augmented `tools/call`, polls `tasks/get` until terminal, and reinterprets `tasks/result` as a normal `CallToolResult`. Fields:
+  - `default_ttl: timedelta | None` — forwarded to the server as `params.task.ttl` (milliseconds). When `None`, the server's default applies.
+  - `cancel_remote_task_on_local_cancellation: bool = True` — on local `CancelledError`, spawn a best-effort `tasks/cancel` before re-raising.
+- **Permissive fallback**: servers that ignore the augmentation (return `CallToolResult` directly) or reject the unknown `task` field with `METHOD_NOT_FOUND` / `INVALID_PARAMS` fall back to the plain `session.call_tool(...)` path so legacy servers keep working.
+- **Phase-aware reconnect**: a dropped connection before a `task_id` is known retries the augmented `tools/call`; once a `task_id` exists, only `tasks/get` / `tasks/result` are retried for the same id so reconnects never create duplicate remote tasks.
+
 ### File Access Harness (`_harness/_file_access.py`)
 
 - **`AgentFileStore`** - Abstract async store backing the file-access harness. Implementations expose `write_file`, `read_file`, `delete_file`, `list_files`, `file_exists`, `search_files`, and `create_directory` over forward-slash relative paths.

python/packages/core/agent_framework/__init__.py

@@ -124,7 +124,7 @@ from ._harness._todo import (
     TodoSessionStore,
     TodoStore,
 )
-from ._mcp import MCPStdioTool, MCPStreamableHTTPTool, MCPWebsocketTool
+from ._mcp import MCPStdioTool, MCPStreamableHTTPTool, MCPTaskOptions, MCPWebsocketTool
 from ._middleware import (
     AgentContext,
     AgentMiddleware,
@@ -444,12 +444,13 @@ __all__ = [
     "InlineSkillResource",
     "InlineSkillScript",
     "LocalEvaluator",
-    "MCPStdioTool",
-    "MCPStreamableHTTPTool",
-    "MCPWebsocketTool",
     "MCPSkill",
     "MCPSkillResource",
     "MCPSkillsSource",
+    "MCPStdioTool",
+    "MCPStreamableHTTPTool",
+    "MCPTaskOptions",
+    "MCPWebsocketTool",
     "MemoryContextProvider",
     "MemoryFileStore",
     "MemoryIndexEntry",

python/packages/core/agent_framework/_feature_stage.py

@@ -58,6 +58,7 @@ class ExperimentalFeature(str, Enum):
     FOUNDRY_PREVIEW_TOOLS = "FOUNDRY_PREVIEW_TOOLS"
     FUNCTIONAL_WORKFLOWS = "FUNCTIONAL_WORKFLOWS"
     HARNESS = "HARNESS"
+    MCP_LONG_RUNNING_TASKS = "MCP_LONG_RUNNING_TASKS"
     MCP_SKILLS = "MCP_SKILLS"
     PROGRESSIVE_TOOLS = "PROGRESSIVE_TOOLS"
     SKILLS = "SKILLS"

python/packages/core/agent_framework/_mcp.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import base64
+import contextlib
 import contextvars
 import json
 import logging
@@ -12,12 +13,14 @@ import sys
 from abc import abstractmethod
 from collections.abc import Callable, Collection, Coroutine, Mapping, Sequence
 from contextlib import AsyncExitStack, _AsyncGeneratorContextManager  # type: ignore
+from dataclasses import dataclass
 from datetime import timedelta
 from functools import partial
 from typing import TYPE_CHECKING, Any, Literal, TypedDict, cast
 
 from opentelemetry import propagate
 
+from ._feature_stage import ExperimentalFeature, experimental
 from ._tools import FunctionTool
 from ._types import (
     ChatOptions,
@@ -149,6 +152,40 @@ def _url_origin(url: Any) -> tuple[str, str, int | None]:
     return (url.scheme, url.host or "", port)
 
 
+# Internal polling bounds for MCP long-running tasks. Not user-tunable today;
+# promote to MCPTaskOptions if a concrete need arises.
+_MCP_TASK_MIN_POLL_INTERVAL = timedelta(milliseconds=500)
+_MCP_TASK_MAX_POLL_INTERVAL = timedelta(seconds=5)
+_MCP_TASK_CANCEL_TIMEOUT = timedelta(seconds=5)
+_MCP_TASK_TERMINAL_STATUSES: frozenset[str] = frozenset({"completed", "failed", "cancelled", "input_required"})
+
+
+@experimental(feature_id=ExperimentalFeature.MCP_LONG_RUNNING_TASKS)
+@dataclass
+class MCPTaskOptions:
+    """Options controlling how MCPTool drives the MCP long-running task lifecycle.
+
+    When an MCP server advertises a tool with ``execution.taskSupport == "required"``,
+    the framework transparently drives the SEP-2663 ``tools/call`` → ``tasks/get``
+    (polled) → ``tasks/result`` lifecycle so the agent sees a normal tool result.
+
+    Attributes:
+        default_ttl: Optional default time-to-live forwarded to the server as
+            ``params.task.ttl`` (milliseconds, integer). When ``None``, the server
+            applies its own default. Must be non-negative if set.
+        cancel_remote_task_on_local_cancellation: If True (default), a local
+            cancellation of the awaiting coroutine triggers a best-effort
+            ``tasks/cancel`` on the server before re-raising ``CancelledError``.
+    """
+
+    default_ttl: timedelta | None = None
+    cancel_remote_task_on_local_cancellation: bool = True
+
+    def __post_init__(self) -> None:
+        if self.default_ttl is not None and self.default_ttl.total_seconds() < 0:
+            raise ValueError("MCPTaskOptions.default_ttl must be non-negative.")
+
+
 def streamable_http_client(*args: Any, **kwargs: Any) -> _AsyncGeneratorContextManager[Any, None]:
     """Lazily import the MCP streamable HTTP transport."""
     try:
@@ -217,6 +254,7 @@ class MCPTool:
         request_timeout: int | None = None,
         client: SupportsChatGetResponse | None = None,
         additional_properties: dict[str, Any] | None = None,
+        task_options: MCPTaskOptions | None = None,
     ) -> None:
         """Initialize the MCP Tool base.
 
@@ -248,6 +286,9 @@ class MCPTool:
             request_timeout: Timeout in seconds for MCP requests.
             client: A chat client for sampling callbacks.
             additional_properties: Additional properties for the tool.
+            task_options: Options controlling how long-running MCP tasks are driven for
+                tools that advertise ``execution.taskSupport == "required"``. When ``None``,
+                the defaults from :class:`MCPTaskOptions` are used.
         """
         self.name = name
         self.description = description or ""
@@ -259,6 +300,10 @@ class MCPTool:
         self.parse_tool_results = parse_tool_results
         self.load_prompts_flag = load_prompts
         self.parse_prompt_results = parse_prompt_results
+        # Defer constructing the default MCPTaskOptions so the experimental warning
+        # only fires when LRO is actually engaged (lazy-resolved by _effective_task_options).
+        self._task_options_explicit: MCPTaskOptions | None = task_options
+        self._task_options_default: MCPTaskOptions | None = None
         self._exit_stack = AsyncExitStack()
         self._lifecycle_lock = asyncio.Lock()
         self._lifecycle_request_lock = asyncio.Lock()
@@ -270,6 +315,7 @@ class MCPTool:
         self.client = client
         self._functions: list[FunctionTool] = []
         self._tool_call_meta_by_name: dict[str, dict[str, Any]] = {}
+        self._tool_task_support_by_name: dict[str, str] = {}
         self.is_connected: bool = False
         self._tools_loaded: bool = False
         self._prompts_loaded: bool = False
@@ -1131,6 +1177,7 @@ class MCPTool:
         # Track existing function names to prevent duplicates
         existing_names = {func.name for func in self._functions}
         tool_call_meta_by_name: dict[str, dict[str, Any]] = {}
+        tool_task_support_by_name: dict[str, str] = {}
 
         params: types.PaginatedRequestParams | None = None
         while True:
@@ -1168,6 +1215,10 @@ class MCPTool:
                 if tool.meta is not None:
                     tool_call_meta_by_name[tool.name] = dict(tool.meta)
 
+                task_support = getattr(getattr(tool, "execution", None), "taskSupport", None)
+                if task_support is not None:
+                    tool_task_support_by_name[tool.name] = task_support
+
                 normalized_name = _normalize_mcp_name(tool.name)
                 local_name = _build_prefixed_mcp_name(normalized_name, self.tool_name_prefix)
 
@@ -1216,6 +1267,7 @@ class MCPTool:
             params = types.PaginatedRequestParams(cursor=tool_list.nextCursor)
 
         self._tool_call_meta_by_name = tool_call_meta_by_name
+        self._tool_task_support_by_name = tool_task_support_by_name
 
     async def _close_on_owner(self) -> None:
         # Cancel any pending reload tasks before tearing down the session.
@@ -1292,6 +1344,29 @@ class MCPTool:
                     inner_exception=ex,
                 ) from ex
 
+    def _effective_task_options(self) -> "MCPTaskOptions":
+        """Return the effective MCPTaskOptions, lazily constructing defaults on first use.
+
+        Defers the implicit ``MCPTaskOptions()`` so the experimental warning only
+        fires when LRO is actually engaged (server advertises ``taskSupport=required``).
+        """
+        explicit = self._task_options_explicit
+        if explicit is not None:
+            return explicit
+        if self._task_options_default is None:
+            self._task_options_default = MCPTaskOptions()
+        return self._task_options_default
+
+    @property
+    def task_options(self) -> "MCPTaskOptions":
+        """The effective MCPTaskOptions for this tool (lazy defaults)."""
+        return self._effective_task_options()
+
+    @task_options.setter
+    def task_options(self, value: "MCPTaskOptions | None") -> None:
+        self._task_options_explicit = value
+        self._task_options_default = None
+
     async def call_tool(self, tool_name: str, **kwargs: Any) -> str | list[Content]:
         """Call a tool with the given arguments.
 
@@ -1322,47 +1397,12 @@ class MCPTool:
                 "Tools are not loaded for this server, please set load_tools=True in the constructor."
             )
 
-        raw_user_meta: object | None = kwargs.get("_meta")
-        user_meta: dict[str, Any] | None = None
-        if raw_user_meta is not None and not isinstance(raw_user_meta, dict):
-            raise ToolExecutionException("MCP tool metadata provided via _meta must be a dict.")
-        if isinstance(raw_user_meta, dict):
-            raw_user_meta_dict = cast(Mapping[object, object], raw_user_meta)
-            user_meta = {}
-            for key, value in raw_user_meta_dict.items():
-                if not isinstance(key, str):
-                    raise ToolExecutionException("MCP tool metadata provided via _meta must use string keys.")
-                user_meta[key] = value
+        # Tools advertising taskSupport == "required" cannot complete via plain tools/call;
+        # route through the long-running task lifecycle transparently.
+        if self._tool_task_support_by_name.get(tool_name) == "required":
+            return await self.call_tool_as_task(tool_name, **kwargs)
 
-        # Filter out framework kwargs that cannot be serialized by the MCP SDK.
-        # These are internal objects passed through the function invocation pipeline
-        # that should not be forwarded to external MCP servers.
-        # conversation_id is an internal tracking ID used by services like Azure AI.
-        # options contains metadata/store used by AG-UI for Azure AI client requirements.
-        # response_format is a Pydantic model class used for structured output (not serializable).
-        filtered_kwargs = {
-            k: v
-            for k, v in kwargs.items()
-            if k
-            not in {
-                "chat_options",
-                "tools",
-                "tool_choice",
-                "session",
-                "thread",
-                "conversation_id",
-                "options",
-                "response_format",
-                "_meta",
-            }
-        }
-
-        # Some MCP proxies require their tools/list metadata to be echoed on tools/call.
-        tool_meta = self._tool_call_meta_by_name.get(tool_name)
-        request_meta = dict(tool_meta) if tool_meta is not None else None
-        if user_meta is not None:
-            request_meta = {**(request_meta or {}), **user_meta}
-        meta = _inject_otel_into_mcp_meta(request_meta)
+        filtered_kwargs, meta = self._prepare_call_kwargs(tool_name, kwargs)
 
         parser = self.parse_tool_results or self._parse_tool_result_from_mcp
         # Try the operation, reconnecting once if the connection is closed
@@ -1411,6 +1451,387 @@ class MCPTool:
                 raise ToolExecutionException(f"Failed to call tool '{tool_name}'.", inner_exception=ex) from ex
         raise ToolExecutionException(f"Failed to call tool '{tool_name}' after retries.")
 
+    def _prepare_call_kwargs(
+        self, tool_name: str, kwargs: dict[str, Any]
+    ) -> tuple[dict[str, Any], dict[str, Any] | None]:
+        """Filter framework-only kwargs and build the merged MCP request metadata."""
+        raw_user_meta: object | None = kwargs.get("_meta")
+        user_meta: dict[str, Any] | None = None
+        if raw_user_meta is not None and not isinstance(raw_user_meta, dict):
+            raise ToolExecutionException("MCP tool metadata provided via _meta must be a dict.")
+        if isinstance(raw_user_meta, dict):
+            raw_user_meta_dict = cast(Mapping[object, object], raw_user_meta)
+            user_meta = {}
+            for key, value in raw_user_meta_dict.items():
+                if not isinstance(key, str):
+                    raise ToolExecutionException("MCP tool metadata provided via _meta must use string keys.")
+                user_meta[key] = value
+
+        # Filter out framework kwargs that cannot be serialized by the MCP SDK.
+        # These are internal objects passed through the function invocation pipeline
+        # that should not be forwarded to external MCP servers.
+        # conversation_id is an internal tracking ID used by services like Azure AI.
+        # options contains metadata/store used by AG-UI for Azure AI client requirements.
+        # response_format is a Pydantic model class used for structured output (not serializable).
+        filtered_kwargs = {
+            k: v
+            for k, v in kwargs.items()
+            if k
+            not in {
+                "chat_options",
+                "tools",
+                "tool_choice",
+                "session",
+                "thread",
+                "conversation_id",
+                "options",
+                "response_format",
+                "_meta",
+            }
+        }
+
+        # Some MCP proxies require their tools/list metadata to be echoed on tools/call.
+        tool_meta = self._tool_call_meta_by_name.get(tool_name)
+        request_meta = dict(tool_meta) if tool_meta is not None else None
+        if user_meta is not None:
+            request_meta = {**(request_meta or {}), **user_meta}
+        meta = _inject_otel_into_mcp_meta(request_meta)
+        return filtered_kwargs, meta
+
+    async def call_tool_as_task(self, tool_name: str, **kwargs: Any) -> str | list[Content]:
+        """Call an MCP tool via the long-running task lifecycle (SEP-2663).
+
+        Issues an augmented ``tools/call`` with ``params.task`` set from
+        ``self.task_options``, then polls ``tasks/get`` until the server reports a
+        terminal status. On ``completed`` the payload is fetched via ``tasks/result``,
+        validated as a ``CallToolResult`` and parsed identically to :meth:`call_tool`.
+
+        Local cancellation triggers a best-effort ``tasks/cancel`` (controlled by
+        :attr:`MCPTaskOptions.cancel_remote_task_on_local_cancellation`) before
+        ``asyncio.CancelledError`` is re-raised.
+
+        Args:
+            tool_name: The remote MCP tool name.
+
+        Keyword Args:
+            kwargs: Arguments forwarded to the tool. See :meth:`call_tool` for the
+                framework kwargs that are filtered out.
+
+        Returns:
+            A list of Content items (or a string when a custom ``parse_tool_results``
+            callback is configured).
+        """
+        from anyio import ClosedResourceError
+        from mcp.shared.exceptions import McpError
+
+        if not self.load_tools_flag:
+            raise ToolExecutionException(
+                "Tools are not loaded for this server, please set load_tools=True in the constructor."
+            )
+
+        filtered_kwargs, meta = self._prepare_call_kwargs(tool_name, kwargs)
+        parser = self.parse_tool_results or self._parse_tool_result_from_mcp
+
+        # Phase 1: issue augmented tools/call. Do NOT retry on connection loss here:
+        # the server may have accepted the request and created a task before the
+        # response was lost, so retrying could start the long-running operation twice.
+        # Reconnect-and-retry is only safe after task_id is known (phase 2).
+        try:
+            task_id, fallback_result = await self._call_tool_as_task_create(tool_name, filtered_kwargs, meta)
+        except (ClosedResourceError, McpError) as ex:
+            if not self._is_connection_lost(ex):
+                error_message = ex.error.message if isinstance(ex, McpError) else str(ex)
+                raise ToolExecutionException(error_message, inner_exception=ex) from ex
+            raise ToolExecutionException(
+                f"Failed to call tool '{tool_name}' - connection lost; task state unknown.",
+                inner_exception=ex,
+            ) from ex
+        except ToolExecutionException:
+            raise
+        except Exception as ex:
+            raise ToolExecutionException(f"Failed to call tool '{tool_name}'.", inner_exception=ex) from ex
+
+        # Server returned a CallToolResult (no task created) or fell back to plain tools/call.
+        if fallback_result is not None:
+            if fallback_result.isError:
+                parsed = parser(fallback_result)
+                text = (
+                    "\n".join(c.text for c in parsed if c.type == "text" and c.text)
+                    if isinstance(parsed, list)
+                    else str(parsed)
+                )
+                raise ToolExecutionException(text or str(parsed))
+            return parser(fallback_result)
+
+        assert task_id is not None  # noqa: S101 - protected by the branch above
+
+        # Phase 2: poll until terminal status, then fetch payload. Never re-issue tools/call
+        # past this point; reconnect-and-retry only against the same task_id.
+        try:
+            terminal = await self._poll_task_until_terminal(task_id)
+            return await self._handle_terminal_task(tool_name, task_id, terminal, parser)
+        except asyncio.CancelledError:
+            if self._effective_task_options().cancel_remote_task_on_local_cancellation:
+                self._spawn_best_effort_cancel(task_id)
+            raise
+
+    async def _call_tool_as_task_create(
+        self, tool_name: str, arguments: dict[str, Any], meta: dict[str, Any] | None
+    ) -> tuple[str | None, types.CallToolResult | None]:
+        """Send the augmented tools/call.
+
+        Returns ``(task_id, None)`` when the server created a task,
+        ``(None, CallToolResult)`` when it returned a non-task result, falling back
+        to plain ``tools/call`` if the server rejects the ``task`` field outright.
+        """
+        from mcp import types
+        from mcp.shared.exceptions import McpError
+        from pydantic import ValidationError
+
+        opts = self._effective_task_options()
+        ttl_ms: int | None = None
+        if opts.default_ttl is not None:
+            ttl_ms = int(opts.default_ttl.total_seconds() * 1000)
+        # Always send TaskMetadata to mark the call as task-augmented; ttl may be omitted.
+        task_metadata = types.TaskMetadata(ttl=ttl_ms)
+
+        request_meta = types.RequestParams.Meta(**meta) if meta else None
+        params = types.CallToolRequestParams(
+            name=tool_name,
+            arguments=arguments,
+            task=task_metadata,
+            _meta=request_meta,  # type: ignore[call-arg]
+        )
+        request = types.ClientRequest(types.CallToolRequest(params=params))
+
+        # Use the lenient Result type so we can extract the task_id even when
+        # the strict CreateTaskResult schema rejects the payload (the MCP Python
+        # SDK requires Task.ttl, but servers may legitimately omit it).
+        try:
+            lenient = await self.session.send_request(  # type: ignore[union-attr]
+                request,
+                types.Result,
+            )
+        except McpError as ex:
+            if ex.error.code not in (types.METHOD_NOT_FOUND, types.INVALID_PARAMS):
+                raise
+            logger.debug(
+                "Server rejected augmented tools/call for '%s' (code=%s); falling back.",
+                tool_name,
+                ex.error.code,
+            )
+            fallback = await self.session.call_tool(tool_name, arguments=arguments, meta=meta)  # type: ignore[union-attr]
+            return None, fallback
+
+        # Inspect the raw payload: a CreateTaskResult carries `task.taskId`;
+        # a legacy CallToolResult carries `content` and/or `isError`.
+        raw: dict[str, Any] = lenient.model_dump(by_alias=True, exclude_none=True)
+        raw.pop("_meta", None)
+
+        task_field = raw.get("task")
+        if isinstance(task_field, dict):
+            task_id_val = cast(dict[str, Any], task_field).get("taskId")
+            if isinstance(task_id_val, str):
+                return task_id_val, None
+
+        try:
+            legacy = types.CallToolResult.model_validate(raw)
+        except ValidationError:
+            logger.debug(
+                "Augmented tools/call for '%s' returned a non-CreateTaskResult/non-CallToolResult payload; "
+                "falling back to plain tools/call.",
+                tool_name,
+            )
+            fallback = await self.session.call_tool(tool_name, arguments=arguments, meta=meta)  # type: ignore[union-attr]
+            return None, fallback
+
+        return None, legacy
+
+    async def _poll_task_until_terminal(self, task_id: str) -> types.GetTaskResult:
+        """Poll ``tasks/get`` until the task reaches a terminal status."""
+        from anyio import ClosedResourceError
+        from mcp import types
+        from mcp.shared.exceptions import McpError
+
+        while True:
+            for attempt in range(2):
+                try:
+                    request = types.ClientRequest(
+                        types.GetTaskRequest(params=types.GetTaskRequestParams(taskId=task_id))
+                    )
+                    # Use lenient Result then coerce: GetTaskResult.ttl is required by
+                    # schema but servers may legitimately omit it.
+                    lenient = await self.session.send_request(request, types.Result)  # type: ignore[union-attr]
+                    snapshot = self._coerce_get_task_result(lenient, task_id)
+                    break
+                except (ClosedResourceError, McpError) as ex:
+                    if not self._is_connection_lost(ex):
+                        error_message = ex.error.message if isinstance(ex, McpError) else str(ex)
+                        raise ToolExecutionException(error_message, inner_exception=ex) from ex
+                    if attempt == 0:
+                        logger.info("MCP connection lost during tasks/get; reconnecting (task_id=%s).", task_id)
+                        try:
+                            await self.connect(reset=True)
+                            continue
+                        except Exception as reconn_ex:
+                            raise ToolExecutionException(
+                                "Failed to reconnect to MCP server.",
+                                inner_exception=reconn_ex,
+                            ) from reconn_ex
+                    raise ToolExecutionException(
+                        f"MCP connection lost; task state unknown (task_id={task_id}).",
+                        inner_exception=ex,
+                    ) from ex
+            else:  # pragma: no cover - defensive
+                raise ToolExecutionException(f"Failed to poll task '{task_id}'.")
+
+            if snapshot.status in _MCP_TASK_TERMINAL_STATUSES:
+                return snapshot
+
+            await asyncio.sleep(self._compute_poll_delay(snapshot.pollInterval).total_seconds())
+
+    @staticmethod
+    def _coerce_get_task_result(lenient: "types.Result", task_id: str) -> "types.GetTaskResult":
+        """Coerce a lenient Result into GetTaskResult, defaulting ``ttl`` when absent."""
+        from mcp import types
+
+        raw = lenient.model_dump(by_alias=True, exclude_none=True)
+        raw.pop("_meta", None)
+        raw.setdefault("ttl", None)
+        try:
+            return types.GetTaskResult.model_validate(raw)
+        except Exception as ex:
+            raise ToolExecutionException(
+                f"MCP server returned a malformed tasks/get response for task '{task_id}'.",
+                inner_exception=ex,
+            ) from ex
+
+    @staticmethod
+    def _compute_poll_delay(server_interval_ms: int | None) -> timedelta:
+        """Clamp the server-suggested poll interval to ``[min, max]``."""
+        if server_interval_ms is None or server_interval_ms <= 0:
+            return _MCP_TASK_MIN_POLL_INTERVAL
+        suggested = timedelta(milliseconds=server_interval_ms)
+        if suggested < _MCP_TASK_MIN_POLL_INTERVAL:
+            return _MCP_TASK_MIN_POLL_INTERVAL
+        if suggested > _MCP_TASK_MAX_POLL_INTERVAL:
+            return _MCP_TASK_MAX_POLL_INTERVAL
+        return suggested
+
+    async def _handle_terminal_task(
+        self,
+        tool_name: str,
+        task_id: str,
+        snapshot: types.GetTaskResult,
+        parser: Callable[[types.CallToolResult], str | list[Content]],
+    ) -> str | list[Content]:
+        """Map a terminal task snapshot to either a parsed result or an exception."""
+        status = snapshot.status
+        if status == "completed":
+            payload = await self._fetch_task_result(task_id)
+            if payload.isError:
+                parsed = parser(payload)
+                text = (
+                    "\n".join(c.text for c in parsed if c.type == "text" and c.text)
+                    if isinstance(parsed, list)
+                    else str(parsed)
+                )
+                raise ToolExecutionException(text or str(parsed))
+            return parser(payload)
+
+        # Non-completed terminal statuses surface as ToolExecutionException so the
+        # function-calling loop sees a normal failure for tool_name.
+        message = snapshot.statusMessage or f"MCP task ended with status '{status}'."
+        if status == "input_required":
+            # Spec-non-terminal; treated as terminal here because the framework does
+            # not implement the interactive input flow.
+            message = snapshot.statusMessage or "MCP task requires additional input and cannot continue."
+        raise ToolExecutionException(f"Tool '{tool_name}' task {status}: {message}")
+
+    async def _fetch_task_result(self, task_id: str) -> types.CallToolResult:
+        """Send ``tasks/result`` and reinterpret the open-typed payload as a CallToolResult."""
+        from anyio import ClosedResourceError
+        from mcp import types
+        from mcp.shared.exceptions import McpError
+        from pydantic import ValidationError
+
+        for attempt in range(2):
+            try:
+                request = types.ClientRequest(
+                    types.GetTaskPayloadRequest(params=types.GetTaskPayloadRequestParams(taskId=task_id))
+                )
+                payload = await self.session.send_request(  # type: ignore[union-attr]
+                    request, types.GetTaskPayloadResult
+                )
+                break
+            except (ClosedResourceError, McpError) as ex:
+                if not self._is_connection_lost(ex):
+                    error_message = ex.error.message if isinstance(ex, McpError) else str(ex)
+                    raise ToolExecutionException(error_message, inner_exception=ex) from ex
+                if attempt == 0:
+                    logger.info("MCP connection lost during tasks/result; reconnecting (task_id=%s).", task_id)
+                    try:
+                        await self.connect(reset=True)
+                        continue
+                    except Exception as reconn_ex:
+                        raise ToolExecutionException(
+                            "Failed to reconnect to MCP server.",
+                            inner_exception=reconn_ex,
+                        ) from reconn_ex
+                raise ToolExecutionException(
+                    f"MCP connection lost; task state unknown (task_id={task_id}).",
+                    inner_exception=ex,
+                ) from ex
+        else:  # pragma: no cover - defensive
+            raise ToolExecutionException(f"Failed to fetch result for task '{task_id}'.")
+
+        # GetTaskPayloadResult carries the tool result via extra fields; reinterpret as CallToolResult.
+        payload_dict = payload.model_dump(by_alias=True, exclude_none=True)
+        payload_dict.pop("_meta", None)
+        try:
+            return types.CallToolResult.model_validate(payload_dict)
+        except ValidationError as ex:
+            raise ToolExecutionException(
+                f"MCP task '{task_id}' result payload could not be parsed as a CallToolResult."
+            ) from ex
+
+    def _spawn_best_effort_cancel(self, task_id: str) -> None:
+        """Fire-and-forget ``tasks/cancel`` so local cancellation propagates server-side."""
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            return
+        cancel_task = loop.create_task(self._try_cancel_task(task_id))
+        # Reuse pending-reload bookkeeping so close-on-owner waits/cancels these too.
+        self._pending_reload_tasks.add(cancel_task)
+        cancel_task.add_done_callback(self._pending_reload_tasks.discard)
+
+    async def _try_cancel_task(self, task_id: str) -> None:
+        """Send ``tasks/cancel`` swallowing every failure; bounded by ``_MCP_TASK_CANCEL_TIMEOUT``."""
+        from mcp import types
+
+        async def _send() -> None:
+            request = types.ClientRequest(types.CancelTaskRequest(params=types.CancelTaskRequestParams(taskId=task_id)))
+            try:
+                await self.session.send_request(request, types.CancelTaskResult)  # type: ignore[union-attr]
+            except Exception:
+                logger.debug("Best-effort tasks/cancel for '%s' failed.", task_id, exc_info=True)
+
+        with contextlib.suppress(Exception, asyncio.CancelledError, asyncio.TimeoutError):
+            await asyncio.wait_for(_send(), timeout=_MCP_TASK_CANCEL_TIMEOUT.total_seconds())
+
+    @staticmethod
+    def _is_connection_lost(ex: BaseException) -> bool:
+        """Return True if *ex* indicates the MCP transport was torn down."""
+        from anyio import ClosedResourceError
+        from mcp.shared.exceptions import McpError
+
+        if isinstance(ex, ClosedResourceError):
+            return True
+        if isinstance(ex, McpError):
+            return "session terminated" in ex.error.message.lower()
+        return False
+
     async def get_prompt(self, prompt_name: str, **kwargs: Any) -> str:
         """Call a prompt with the given arguments.
 
@@ -1554,6 +1975,7 @@ class MCPStdioTool(MCPTool):
         encoding: str | None = None,
         client: SupportsChatGetResponse | None = None,
         additional_properties: dict[str, Any] | None = None,
+        task_options: MCPTaskOptions | None = None,
         **kwargs: Any,
     ) -> None:
         """Initialize the MCP stdio tool.
@@ -1598,6 +2020,8 @@ class MCPStdioTool(MCPTool):
             env: The environment variables to set for the command.
             encoding: The encoding to use for the command output.
             client: The chat client to use for sampling.
+            task_options: Options for tools that advertise
+                ``execution.taskSupport == "required"``. See :class:`MCPTaskOptions`.
             kwargs: Any extra arguments to pass to the stdio client.
         """
         super().__init__(
@@ -1614,6 +2038,7 @@ class MCPStdioTool(MCPTool):
             load_prompts=load_prompts,
             parse_prompt_results=parse_prompt_results,
             request_timeout=request_timeout,
+            task_options=task_options,
         )
         self.command = command
         self.args = args or []
@@ -1687,6 +2112,7 @@ class MCPStreamableHTTPTool(MCPTool):
         additional_properties: dict[str, Any] | None = None,
         http_client: AsyncClient | None = None,
         header_provider: Callable[[dict[str, Any]], dict[str, str]] | None = None,
+        task_options: MCPTaskOptions | None = None,
         **kwargs: Any,
     ) -> None:
         """Initialize the MCP streamable HTTP tool.
@@ -1739,6 +2165,8 @@ class MCPStreamableHTTPTool(MCPTool):
                 of HTTP headers to inject into every outbound request to the MCP server.
                 Use this to forward per-request context (e.g. authentication tokens set in
                 agent middleware) without creating a separate ``httpx.AsyncClient``.
+            task_options: Options for tools that advertise
+                ``execution.taskSupport == "required"``. See :class:`MCPTaskOptions`.
             kwargs: Additional keyword arguments (accepted for backward compatibility but not used).
         """
         super().__init__(
@@ -1755,6 +2183,7 @@ class MCPStreamableHTTPTool(MCPTool):
             load_prompts=load_prompts,
             parse_prompt_results=parse_prompt_results,
             request_timeout=request_timeout,
+            task_options=task_options,
         )
         self.url = url
         self.terminate_on_close = terminate_on_close
@@ -1862,6 +2291,7 @@ class MCPWebsocketTool(MCPTool):
         allowed_tools: Collection[str] | None = None,
         client: SupportsChatGetResponse | None = None,
         additional_properties: dict[str, Any] | None = None,
+        task_options: MCPTaskOptions | None = None,
         **kwargs: Any,
     ) -> None:
         """Initialize the MCP WebSocket tool.
@@ -1904,6 +2334,8 @@ class MCPWebsocketTool(MCPTool):
             allowed_tools: A list of tools that are allowed to use this tool.
             additional_properties: Additional properties.
             client: The chat client to use for sampling.
+            task_options: Options for tools that advertise
+                ``execution.taskSupport == "required"``. See :class:`MCPTaskOptions`.
             kwargs: Any extra arguments to pass to the WebSocket client.
         """
         super().__init__(
@@ -1920,6 +2352,7 @@ class MCPWebsocketTool(MCPTool):
             load_prompts=load_prompts,
             parse_prompt_results=parse_prompt_results,
             request_timeout=request_timeout,
+            task_options=task_options,
         )
         self.url = url
         self._client_kwargs = kwargs

python/packages/core/tests/core/test_mcp.py

@@ -4969,3 +4969,588 @@ async def test_mcp_streamable_http_tool_header_provider_via_invoke_with_context(
 
 
 # endregion
+
+
+# region: MCP long-running task (SEP-2663) tests
+
+
+def _utc_now() -> Any:
+    from datetime import datetime, timezone
+
+    return datetime.now(timezone.utc)
+
+
+def _make_task_snapshot(
+    *,
+    task_id: str = "task-1",
+    status: str = "working",
+    status_message: str | None = None,
+    poll_interval_ms: int | None = None,
+) -> types.GetTaskResult:
+    now = _utc_now()
+    return types.GetTaskResult(
+        taskId=task_id,
+        status=status,  # type: ignore[arg-type]
+        statusMessage=status_message,
+        createdAt=now,
+        lastUpdatedAt=now,
+        ttl=None,
+        pollInterval=poll_interval_ms,
+    )
+
+
+def _make_create_task_result(task_id: str = "task-1") -> types.CreateTaskResult:
+    now = _utc_now()
+    return types.CreateTaskResult(
+        task=types.Task(
+            taskId=task_id,
+            status="working",
+            statusMessage=None,
+            createdAt=now,
+            lastUpdatedAt=now,
+            ttl=None,
+        )
+    )
+
+
+def _make_payload(text: str = "done!", is_error: bool = False) -> types.GetTaskPayloadResult:
+    return types.GetTaskPayloadResult.model_validate({
+        "content": [{"type": "text", "text": text}],
+        "isError": is_error,
+    })
+
+
+def _make_task_tool(
+    tool_name: str = "slow_op",
+    *,
+    task_support: str | None = "required",
+    task_options: Any = None,
+) -> MCPTool:
+    from agent_framework import MCPTaskOptions
+
+    tool = MCPTool(
+        name="lro",
+        task_options=task_options if task_options is not None else MCPTaskOptions(),
+    )
+    tool.session = AsyncMock(spec=ClientSession)
+    if task_support is not None:
+        tool._tool_task_support_by_name[tool_name] = task_support
+    return tool
+
+
+def _send_request_dispatcher(*responses_by_method: tuple[str, Any]) -> Any:
+    """Build a send_request side_effect that returns responses keyed by request method.
+
+    Each tuple is ``(method_name, response_or_exception_or_callable)``. The dispatcher
+    advances a per-method queue on every call. A callable response is invoked with no
+    args so tests can raise exceptions deterministically.
+    """
+    from collections import defaultdict
+
+    queues: dict[str, list[Any]] = defaultdict(list)
+    for method, response in responses_by_method:
+        queues[method].append(response)
+
+    async def _dispatch(request: Any, _result_type: Any, *_args: Any, **_kw: Any) -> Any:
+        method = getattr(request.root, "method", None) or getattr(request, "method", None)
+        queue = queues.get(method)
+        if not queue:
+            raise AssertionError(f"No mocked send_request response for method '{method}'.")
+        item = queue.pop(0)
+        if callable(item):
+            return item()
+        if isinstance(item, BaseException):
+            raise item
+        return item
+
+    return _dispatch
+
+
+async def test_task_options_defaults_are_sane() -> None:
+    from agent_framework import MCPTaskOptions
+
+    opts = MCPTaskOptions()
+    assert opts.default_ttl is None
+    assert opts.cancel_remote_task_on_local_cancellation is True
+
+
+async def test_task_options_rejects_negative_default_ttl() -> None:
+    from datetime import timedelta
+
+    from agent_framework import MCPTaskOptions
+
+    with pytest.raises(ValueError, match="non-negative"):
+        MCPTaskOptions(default_ttl=timedelta(seconds=-1))
+
+
+async def test_load_tools_captures_task_support() -> None:
+    tool = MCPTool(name="lro")
+    tool.session = AsyncMock()
+    tool.load_tools_flag = True
+
+    page = Mock()
+    page.tools = [
+        types.Tool(
+            name="slow_op",
+            description="slow",
+            inputSchema={"type": "object", "properties": {}},
+            execution=types.ToolExecution(taskSupport="required"),
+        ),
+        types.Tool(
+            name="fast_op",
+            description="fast",
+            inputSchema={"type": "object", "properties": {}},
+        ),
+    ]
+    page.nextCursor = None
+    tool.session.list_tools = AsyncMock(return_value=page)
+
+    await tool.load_tools()
+
+    assert tool._tool_task_support_by_name == {"slow_op": "required"}
+
+
+async def test_call_tool_routes_required_through_task_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None:
+    from agent_framework import _mcp as _mcp_module
+
+    monkeypatch.setattr(_mcp_module, "_MCP_TASK_MIN_POLL_INTERVAL", _mcp_module.timedelta(milliseconds=1))
+
+    tool = _make_task_tool()
+    tool.session.send_request = AsyncMock(  # type: ignore[union-attr]
+        side_effect=_send_request_dispatcher(
+            ("tools/call", _make_create_task_result()),
+            ("tasks/get", _make_task_snapshot(status="working")),
+            ("tasks/get", _make_task_snapshot(status="completed")),
+            ("tasks/result", _make_payload("hello task")),
+        )
+    )
+
+    result = await tool.call_tool("slow_op", x=1)
+
+    assert _mcp_result_to_text(result) == "hello task"
+    # Plain session.call_tool must NOT be used for required tools.
+    tool.session.call_tool.assert_not_called()  # type: ignore[union-attr]
+
+
+async def test_call_tool_as_task_default_ttl_propagates() -> None:
+    from datetime import timedelta
+
+    from agent_framework import MCPTaskOptions
+
+    tool = _make_task_tool(task_options=MCPTaskOptions(default_ttl=timedelta(minutes=7)))
+
+    captured: list[Any] = []
+
+    async def fake_send(request: Any, _result_type: Any, *_a: Any, **_kw: Any) -> Any:
+        captured.append(request)
+        method = request.root.method
+        if method == "tools/call":
+            return _make_create_task_result()
+        if method == "tasks/get":
+            return _make_task_snapshot(status="completed")
+        if method == "tasks/result":
+            return _make_payload("ok")
+        raise AssertionError(method)
+
+    tool.session.send_request = AsyncMock(side_effect=fake_send)  # type: ignore[union-attr]
+
+    await tool.call_tool("slow_op")
+
+    create_req = captured[0]
+    assert create_req.root.method == "tools/call"
+    assert create_req.root.params.task is not None
+    assert create_req.root.params.task.ttl == 7 * 60 * 1000
+
+
+async def test_call_tool_as_task_sends_empty_task_metadata_when_ttl_none() -> None:
+    # Without a TTL we still mark the call as task-augmented (servers require
+    # the `task` field to route through the lifecycle).
+    tool = _make_task_tool()
+
+    captured: list[Any] = []
+
+    async def fake_send(request: Any, _result_type: Any, *_a: Any, **_kw: Any) -> Any:
+        captured.append(request)
+        method = request.root.method
+        if method == "tools/call":
+            return _make_create_task_result()
+        if method == "tasks/get":
+            return _make_task_snapshot(status="completed")
+        if method == "tasks/result":
+            return _make_payload("ok")
+        raise AssertionError(method)
+
+    tool.session.send_request = AsyncMock(side_effect=fake_send)  # type: ignore[union-attr]
+
+    await tool.call_tool("slow_op")
+
+    create_req = captured[0]
+    assert create_req.root.method == "tools/call"
+    assert create_req.root.params.task is not None
+    assert create_req.root.params.task.ttl is None
+
+
+async def test_call_tool_skips_task_path_for_optional_and_forbidden() -> None:
+    for support in ("optional", "forbidden", None):
+        tool = _make_task_tool(task_support=support)
+        tool.session.call_tool = AsyncMock(  # type: ignore[union-attr]
+            return_value=types.CallToolResult(content=[types.TextContent(type="text", text="plain")])
+        )
+        tool.session.send_request = AsyncMock(side_effect=AssertionError("task path should not be used"))  # type: ignore[union-attr]
+
+        result = await tool.call_tool("slow_op")
+        assert _mcp_result_to_text(result) == "plain"
+
+
+async def test_call_tool_as_task_cancelled_status_raises() -> None:
+    tool = _make_task_tool()
+    tool.session.send_request = AsyncMock(  # type: ignore[union-attr]
+        side_effect=_send_request_dispatcher(
+            ("tools/call", _make_create_task_result()),
+            ("tasks/get", _make_task_snapshot(status="cancelled", status_message="server stop")),
+        )
+    )
+
+    with pytest.raises(ToolExecutionException, match="cancelled.*server stop"):
+        await tool.call_tool("slow_op")
+
+
+async def test_call_tool_as_task_failed_status_raises() -> None:
+    tool = _make_task_tool()
+    tool.session.send_request = AsyncMock(  # type: ignore[union-attr]
+        side_effect=_send_request_dispatcher(
+            ("tools/call", _make_create_task_result()),
+            ("tasks/get", _make_task_snapshot(status="failed", status_message="boom")),
+        )
+    )
+
+    with pytest.raises(ToolExecutionException, match="failed.*boom"):
+        await tool.call_tool("slow_op")
+
+
+async def test_call_tool_as_task_input_required_raises() -> None:
+    tool = _make_task_tool()
+    tool.session.send_request = AsyncMock(  # type: ignore[union-attr]
+        side_effect=_send_request_dispatcher(
+            ("tools/call", _make_create_task_result()),
+            ("tasks/get", _make_task_snapshot(status="input_required", status_message="need more")),
+        )
+    )
+
+    with pytest.raises(ToolExecutionException, match="input_required.*need more"):
+        await tool.call_tool("slow_op")
+
+
+async def test_call_tool_as_task_payload_iserror_raises() -> None:
+    tool = _make_task_tool()
+    tool.session.send_request = AsyncMock(  # type: ignore[union-attr]
+        side_effect=_send_request_dispatcher(
+            ("tools/call", _make_create_task_result()),
+            ("tasks/get", _make_task_snapshot(status="completed")),
+            ("tasks/result", _make_payload("payload exploded", is_error=True)),
+        )
+    )
+
+    with pytest.raises(ToolExecutionException, match="payload exploded"):
+        await tool.call_tool("slow_op")
+
+
+async def test_call_tool_as_task_malformed_payload_raises() -> None:
+    tool = _make_task_tool()
+    bad_payload = types.GetTaskPayloadResult.model_validate({"random": "stuff"})
+    tool.session.send_request = AsyncMock(  # type: ignore[union-attr]
+        side_effect=_send_request_dispatcher(
+            ("tools/call", _make_create_task_result(task_id="abc")),
+            ("tasks/get", _make_task_snapshot(task_id="abc", status="completed")),
+            ("tasks/result", bad_payload),
+        )
+    )
+
+    with pytest.raises(ToolExecutionException, match="task 'abc' result payload"):
+        await tool.call_tool("slow_op")
+
+
+async def test_call_tool_as_task_method_not_found_falls_back() -> None:
+    tool = _make_task_tool()
+    tool.session.send_request = AsyncMock(  # type: ignore[union-attr]
+        side_effect=McpError(types.ErrorData(code=types.METHOD_NOT_FOUND, message="no tasks here"))
+    )
+    tool.session.call_tool = AsyncMock(  # type: ignore[union-attr]
+        return_value=types.CallToolResult(content=[types.TextContent(type="text", text="fell back")])
+    )
+
+    result = await tool.call_tool("slow_op")
+
+    assert _mcp_result_to_text(result) == "fell back"
+    tool.session.call_tool.assert_awaited_once()  # type: ignore[union-attr]
+
+
+async def test_call_tool_as_task_invalid_params_falls_back() -> None:
+    tool = _make_task_tool()
+    tool.session.send_request = AsyncMock(  # type: ignore[union-attr]
+        side_effect=McpError(types.ErrorData(code=types.INVALID_PARAMS, message="unknown field"))
+    )
+    tool.session.call_tool = AsyncMock(  # type: ignore[union-attr]
+        return_value=types.CallToolResult(content=[types.TextContent(type="text", text="plain ok")])
+    )
+
+    result = await tool.call_tool("slow_op")
+
+    assert _mcp_result_to_text(result) == "plain ok"
+
+
+async def test_call_tool_as_task_legacy_calltoolresult_response_used_directly() -> None:
+    """Server may ignore augmentation and return CallToolResult; treat it as the result."""
+    # Build a lenient Result whose extras match a CallToolResult shape.
+    legacy_payload = types.Result.model_validate({
+        "content": [{"type": "text", "text": "legacy ok"}],
+        "isError": False,
+    })
+
+    tool = _make_task_tool()
+    tool.session.send_request = AsyncMock(return_value=legacy_payload)  # type: ignore[union-attr]
+
+    result = await tool.call_tool("slow_op")
+
+    assert _mcp_result_to_text(result) == "legacy ok"
+    # Polling must not occur: a single tools/call was enough.
+    assert tool.session.send_request.call_count == 1  # type: ignore[union-attr]
+
+
+async def test_call_tool_as_task_poll_interval_is_clamped(monkeypatch: pytest.MonkeyPatch) -> None:
+    from datetime import timedelta as _td
+
+    from agent_framework import _mcp as _mcp_module
+
+    # Stub asyncio.sleep so we can capture delays without actually sleeping.
+    delays: list[float] = []
+
+    async def fake_sleep(delay: float) -> None:
+        delays.append(delay)
+
+    monkeypatch.setattr(_mcp_module.asyncio, "sleep", fake_sleep)
+
+    tool = _make_task_tool()
+    tool.session.send_request = AsyncMock(  # type: ignore[union-attr]
+        side_effect=_send_request_dispatcher(
+            ("tools/call", _make_create_task_result()),
+            ("tasks/get", _make_task_snapshot(status="working", poll_interval_ms=50)),  # below 500ms min
+            ("tasks/get", _make_task_snapshot(status="working", poll_interval_ms=10_000)),  # above 5s max
+            ("tasks/get", _make_task_snapshot(status="working", poll_interval_ms=None)),  # default to min
+            ("tasks/get", _make_task_snapshot(status="working", poll_interval_ms=0)),  # invalid -> min
+            ("tasks/get", _make_task_snapshot(status="working", poll_interval_ms=2_000)),  # in-band
+            ("tasks/get", _make_task_snapshot(status="completed")),
+            ("tasks/result", _make_payload("ok")),
+        )
+    )
+
+    await tool.call_tool("slow_op")
+
+    expected = [
+        _td(milliseconds=500).total_seconds(),  # clamp up
+        _td(seconds=5).total_seconds(),  # clamp down
+        _td(milliseconds=500).total_seconds(),  # missing -> min
+        _td(milliseconds=500).total_seconds(),  # zero    -> min
+        _td(milliseconds=2_000).total_seconds(),
+    ]
+    assert delays == expected
+
+
+async def test_call_tool_as_task_local_cancellation_fires_remote_cancel(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from agent_framework import _mcp as _mcp_module
+
+    monkeypatch.setattr(_mcp_module, "_MCP_TASK_MIN_POLL_INTERVAL", _mcp_module.timedelta(milliseconds=1))
+
+    tool = _make_task_tool()
+
+    cancel_seen = asyncio.Event()
+    create_seen = asyncio.Event()
+
+    async def fake_send(request: Any, _result_type: Any, *_a: Any, **_kw: Any) -> Any:
+        method = request.root.method
+        if method == "tools/call":
+            create_seen.set()
+            return _make_create_task_result()
+        if method == "tasks/get":
+            await asyncio.sleep(0)
+            return _make_task_snapshot(status="working")
+        if method == "tasks/cancel":
+            cancel_seen.set()
+            return types.CancelTaskResult()
+        raise AssertionError(method)
+
+    tool.session.send_request = AsyncMock(side_effect=fake_send)  # type: ignore[union-attr]
+
+    task = asyncio.create_task(tool.call_tool("slow_op"))
+    await asyncio.wait_for(create_seen.wait(), timeout=1.0)
+    # Let polling iterate a few times.
+    await asyncio.sleep(0.02)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    # Wait for the fire-and-forget cancel to complete.
+    await asyncio.wait_for(cancel_seen.wait(), timeout=1.0)
+    # Drain any tracked background tasks.
+    pending = list(tool._pending_reload_tasks)
+    if pending:
+        await asyncio.gather(*pending, return_exceptions=True)
+
+
+async def test_call_tool_as_task_cancellation_suppressed_when_disabled(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from agent_framework import MCPTaskOptions
+    from agent_framework import _mcp as _mcp_module
+
+    monkeypatch.setattr(_mcp_module, "_MCP_TASK_MIN_POLL_INTERVAL", _mcp_module.timedelta(milliseconds=1))
+
+    tool = _make_task_tool(
+        task_options=MCPTaskOptions(cancel_remote_task_on_local_cancellation=False),
+    )
+
+    cancel_called = False
+    create_seen = asyncio.Event()
+
+    async def fake_send(request: Any, _result_type: Any, *_a: Any, **_kw: Any) -> Any:
+        nonlocal cancel_called
+        method = request.root.method
+        if method == "tools/call":
+            create_seen.set()
+            return _make_create_task_result()
+        if method == "tasks/get":
+            await asyncio.sleep(0)
+            return _make_task_snapshot(status="working")
+        if method == "tasks/cancel":
+            cancel_called = True
+            return types.CancelTaskResult()
+        raise AssertionError(method)
+
+    tool.session.send_request = AsyncMock(side_effect=fake_send)  # type: ignore[union-attr]
+
+    task = asyncio.create_task(tool.call_tool("slow_op"))
+    await asyncio.wait_for(create_seen.wait(), timeout=1.0)
+    await asyncio.sleep(0.02)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    # Let any (incorrect) background work settle, then verify cancel was NOT sent.
+    await asyncio.sleep(0.02)
+    assert cancel_called is False
+
+
+async def test_call_tool_as_task_reconnects_during_poll(monkeypatch: pytest.MonkeyPatch) -> None:
+    from anyio import ClosedResourceError
+
+    from agent_framework import _mcp as _mcp_module
+
+    monkeypatch.setattr(_mcp_module, "_MCP_TASK_MIN_POLL_INTERVAL", _mcp_module.timedelta(milliseconds=1))
+
+    tool = _make_task_tool()
+
+    poll_calls = 0
+
+    async def fake_send(request: Any, _result_type: Any, *_a: Any, **_kw: Any) -> Any:
+        nonlocal poll_calls
+        method = request.root.method
+        if method == "tools/call":
+            return _make_create_task_result(task_id="abc")
+        if method == "tasks/get":
+            poll_calls += 1
+            assert request.root.params.taskId == "abc"
+            if poll_calls == 1:
+                raise ClosedResourceError
+            return _make_task_snapshot(task_id="abc", status="completed")
+        if method == "tasks/result":
+            return _make_payload("recovered")
+        raise AssertionError(method)
+
+    tool.session.send_request = AsyncMock(side_effect=fake_send)  # type: ignore[union-attr]
+
+    reconnect_calls = 0
+
+    async def fake_connect(reset: bool = False) -> None:
+        nonlocal reconnect_calls
+        reconnect_calls += 1
+        assert reset is True
+
+    with patch.object(MCPTool, "connect", side_effect=fake_connect):
+        result = await tool.call_tool("slow_op")
+
+    assert _mcp_result_to_text(result) == "recovered"
+    assert reconnect_calls == 1
+    # Critically, tools/call must NOT be re-issued after task_id is known.
+    assert (
+        sum(
+            1
+            for c in tool.session.send_request.await_args_list  # type: ignore[union-attr]
+            if c.args[0].root.method == "tools/call"
+        )
+        == 1
+    )
+
+
+async def test_call_tool_as_task_second_disconnect_raises_connection_lost(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from anyio import ClosedResourceError
+
+    from agent_framework import _mcp as _mcp_module
+
+    monkeypatch.setattr(_mcp_module, "_MCP_TASK_MIN_POLL_INTERVAL", _mcp_module.timedelta(milliseconds=1))
+
+    tool = _make_task_tool()
+
+    async def fake_send(request: Any, _result_type: Any, *_a: Any, **_kw: Any) -> Any:
+        method = request.root.method
+        if method == "tools/call":
+            return _make_create_task_result(task_id="abc")
+        if method == "tasks/get":
+            raise ClosedResourceError
+        raise AssertionError(method)
+
+    tool.session.send_request = AsyncMock(side_effect=fake_send)  # type: ignore[union-attr]
+
+    with (
+        patch.object(MCPTool, "connect", new=AsyncMock(return_value=None)),
+        pytest.raises(ToolExecutionException, match="task state unknown"),
+    ):
+        await tool.call_tool("slow_op")
+
+
+async def test_call_tool_as_task_create_disconnect_does_not_retry() -> None:
+    """A connection loss during the augmented tools/call must NOT retry.
+
+    Retrying could spawn a duplicate long-running task on the server, because the
+    first request may have been accepted before the response was lost.
+    """
+    from anyio import ClosedResourceError
+
+    tool = _make_task_tool()
+
+    send_calls = 0
+
+    async def fake_send(_request: Any, _result_type: Any, *_a: Any, **_kw: Any) -> Any:
+        nonlocal send_calls
+        send_calls += 1
+        raise ClosedResourceError
+
+    tool.session.send_request = AsyncMock(side_effect=fake_send)  # type: ignore[union-attr]
+
+    reconnect_mock = AsyncMock(return_value=None)
+    with (
+        patch.object(MCPTool, "connect", new=reconnect_mock),
+        pytest.raises(ToolExecutionException, match="task state unknown"),
+    ):
+        await tool.call_tool("slow_op")
+
+    # Exactly one tools/call was issued — the server-side task state is unknown,
+    # so retry is unsafe and must be skipped.
+    assert send_calls == 1
+    reconnect_mock.assert_not_awaited()
+
+
+# endregion

python/samples/02-agents/mcp/README.md

@@ -13,9 +13,12 @@ The Model Context Protocol (MCP) is an open standard for connecting AI agents to
 | **Agent as MCP Server** | [`agent_as_mcp_server.py`](agent_as_mcp_server.py) | Shows how to expose an Agent Framework agent as an MCP server that other AI applications can connect to |
 | **API Key Authentication** | [`mcp_api_key_auth.py`](mcp_api_key_auth.py) | Demonstrates API key authentication with MCP servers using `header_provider`, runtime invocation kwargs, and a command-line API key argument |
 | **GitHub Integration with PAT** | [`mcp_github_pat.py`](mcp_github_pat.py) | Demonstrates connecting to GitHub's MCP server using Personal Access Token (PAT) authentication |
+| **Long-Running Task** | [`mcp_long_running_task.py`](mcp_long_running_task.py) | Demonstrates transparent SEP-2663 long-running task handling for MCP tools that advertise `taskSupport=required`. Self-spawns a stdio MCP child server |
 
 ## Prerequisites
 
+Most samples in this folder use OpenAI:
+
 - `OPENAI_API_KEY` environment variable
 - `OPENAI_CHAT_MODEL` environment variable
 
@@ -23,3 +26,8 @@ Run `mcp_api_key_auth.py` with the MCP API key as the first command-line argumen
 
 For `mcp_github_pat.py`:
 - `GITHUB_PAT` - Your GitHub Personal Access Token (create at https://github.com/settings/tokens)
+
+For `mcp_long_running_task.py` (uses Azure OpenAI via Entra-ID):
+- Run `az login` once
+- `AZURE_OPENAI_ENDPOINT` - your Azure OpenAI resource endpoint, e.g. `https://<resource>.openai.azure.com/`
+- `AZURE_OPENAI_CHAT_MODEL` (or `AZURE_OPENAI_MODEL`) - the deployment name (e.g. `gpt-4o-mini`)

python/samples/02-agents/mcp/mcp_long_running_task.py

@@ -0,0 +1,181 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""
+MCP Long-Running Task (SEP-2663) Example
+
+Demonstrates that ``MCPStdioTool`` transparently drives the MCP long-running
+task lifecycle for tools that advertise ``execution.taskSupport == "required"``.
+The agent observes a single function-call result; the framework handles the
+``tools/call`` → ``tasks/get`` (polled) → ``tasks/result`` sequence in the
+background.
+
+Run it as a single file. The script doubles as both the client and the stdio
+MCP child server (the child branch is selected via ``--server``):
+
+    python mcp_long_running_task.py
+
+Requirements:
+- Azure CLI sign-in (``az login``) — used for Entra-ID auth against Azure OpenAI.
+- ``AZURE_OPENAI_ENDPOINT`` — your Azure OpenAI resource endpoint, e.g.
+  ``https://<resource>.openai.azure.com/``.
+- ``AZURE_OPENAI_CHAT_MODEL`` (or ``AZURE_OPENAI_MODEL``) — the deployment name,
+  e.g. ``gpt-4o-mini``.
+
+This sample uses the lower-level ``mcp.server.lowlevel.Server`` so it can:
+1. Advertise a tool with ``execution=ToolExecution(taskSupport="required")``.
+2. Enable the SDK's experimental task support for the ``tasks/*`` lifecycle.
+"""
+
+import asyncio
+import sys
+from datetime import timedelta
+from typing import Any
+
+from agent_framework import Agent, MCPStdioTool, MCPTaskOptions
+from agent_framework.openai import OpenAIChatClient
+from azure.identity import AzureCliCredential
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+# ---------------------------------------------------------------------------
+# MCP stdio server (child-process branch)
+# ---------------------------------------------------------------------------
+
+
+async def _run_server() -> None:
+    """Run a minimal stdio MCP server exposing one long-running tool."""
+    import mcp.types as types
+    from mcp.server.lowlevel import Server
+    from mcp.server.stdio import stdio_server
+
+    server: Server[Any, Any] = Server("mcp-long-running-task-demo")
+    # Auto-registers handlers for tasks/get, tasks/result, tasks/cancel, tasks/list
+    # backed by an in-memory store.
+    server.experimental.enable_tasks()
+
+    @server.list_tools()
+    async def _list_tools() -> list[types.Tool]:  # pyright: ignore[reportUnusedFunction]
+        return [
+            types.Tool(
+                name="slow_summary",
+                description=(
+                    "Produces a short summary of the supplied text after simulating several seconds of expensive work."
+                ),
+                inputSchema={
+                    "type": "object",
+                    "properties": {
+                        "text": {
+                            "type": "string",
+                            "description": "Text to summarize.",
+                        }
+                    },
+                    "required": ["text"],
+                },
+                # Advertise that this tool MUST be invoked via the task lifecycle.
+                execution=types.ToolExecution(taskSupport="required"),
+            )
+        ]
+
+    @server.call_tool()
+    async def _call_tool(name: str, arguments: dict[str, Any]) -> Any:  # pyright: ignore[reportUnusedFunction]
+        if name != "slow_summary":
+            raise ValueError(f"Unknown tool: {name}")
+
+        ctx = server.request_context
+
+        async def _work(task: Any) -> types.CallToolResult:
+            await task.update_status("Thinking...")
+            await asyncio.sleep(15.0)
+            text: str = (arguments.get("text") or "").strip()
+            words = text.split()
+            preview = " ".join(words[:6]) + ("..." if len(words) > 6 else "")
+            summary = (
+                f"Summarized {len(words)} word(s). First few words: '{preview}'."
+                if words
+                else "No input text was provided."
+            )
+            return types.CallToolResult(
+                content=[types.TextContent(type="text", text=summary)],
+                isError=False,
+            )
+
+        if not ctx.experimental.is_task:
+            # Client invoked the tool without task augmentation. Return a hard
+            # error so a misconfigured client surfaces the problem clearly.
+            return types.CallToolResult(
+                content=[
+                    types.TextContent(
+                        type="text",
+                        text="'slow_summary' must be invoked as a task.",
+                    )
+                ],
+                isError=True,
+            )
+
+        return await ctx.experimental.run_task(_work)
+
+    async with stdio_server() as (read_stream, write_stream):
+        await server.run(read_stream, write_stream, server.create_initialization_options())
+
+
+# ---------------------------------------------------------------------------
+# Agent client (default branch)
+# ---------------------------------------------------------------------------
+
+
+async def _run_client() -> None:
+    mcp_tool = MCPStdioTool(
+        name="LongRunningDemo",
+        description="Demo MCP server exposing a tool that advertises taskSupport=required.",
+        command=sys.executable,
+        args=[__file__, "--server"],
+        # Optional: cap individual tasks at two minutes. The server may apply its
+        # own default if this is omitted.
+        task_options=MCPTaskOptions(default_ttl=timedelta(minutes=2)),
+    )
+
+    async with Agent(
+        client=OpenAIChatClient(credential=AzureCliCredential()),
+        name="LROAgent",
+        instructions=(
+            "You are a helpful assistant. Use the slow_summary tool when the user "
+            "asks for a summary. Wait for the result and present it directly."
+        ),
+        tools=mcp_tool,
+    ) as agent:
+        prompt = (
+            "Please summarize the following text using your slow_summary tool: "
+            "'The Model Context Protocol lets language models talk to external "
+            "tools and resources through a small JSON-RPC surface.'"
+        )
+
+        print("=== run() ===")
+        print(f"User: {prompt}")
+        response = await agent.run(prompt)
+        print(f"Agent: {response.text}\n")
+
+        print("=== run(stream=True) ===")
+        print(f"User: {prompt}")
+        print("Agent: ", end="", flush=True)
+        async for update in agent.run(prompt, stream=True):
+            if update.text:
+                print(update.text, end="", flush=True)
+        print()
+
+
+# ---------------------------------------------------------------------------
+# Entry point
+# ---------------------------------------------------------------------------
+
+
+def main() -> None:
+    if len(sys.argv) > 1 and sys.argv[1] == "--server":
+        asyncio.run(_run_server())
+        return
+    asyncio.run(_run_client())
+
+
+if __name__ == "__main__":
+    main()