feat(cli): expose MCP task settings

libs/cli/generate_schema.py

@@ -23,6 +23,8 @@ from langgraph_cli.schemas import (
     GraphDef,
     HttpConfig,
     IndexConfig,
+    McpConfig,
+    McpTasksConfig,
     SecurityConfig,
     SerdeConfig,
     StoreConfig,
@@ -115,6 +117,8 @@ def add_descriptions_to_schema(schema, cls):
                 AuthConfig,
                 SecurityConfig,
                 HttpConfig,
+                McpConfig,
+                McpTasksConfig,
                 CorsConfig,
                 CacheConfig,
                 ThreadTTLConfig,

libs/cli/langgraph_cli/schemas.py

@@ -439,6 +439,24 @@ class ConfigurableHeaderConfig(TypedDict, total=False):
     """
 
 
+class McpTasksConfig(TypedDict, total=False):
+    """Settings for authenticated MCP elicitation tasks."""
+
+    poll_interval_ms: int
+    """Suggested client polling interval, in milliseconds. Default 1000; range 100–60000."""
+    ttl_minutes: int
+    """Maximum task lifetime, in minutes. Default 1440; range 1–10080."""
+    completed_grace_seconds: int
+    """Result retention after completion, in seconds. Default 300; range 1–86400."""
+    allow_insecure_urls: bool
+    """Allow HTTP elicitation URLs for development. Defaults to False; production URLs require HTTPS."""
+
+
+class McpConfig(TypedDict, total=False):
+    tasks: McpTasksConfig
+    """Polling, retention and URL settings for the MCP Tasks extension."""
+
+
 class HttpConfig(TypedDict, total=False):
     """Configuration for the built-in HTTP server that powers your deployment's routes and endpoints."""
 
@@ -473,6 +491,8 @@ class HttpConfig(TypedDict, total=False):
 
     Default is False.
     """
+    mcp: McpConfig
+    """Optional. Configure the MCP Tasks extension for elicitation over protocol 2026-07-28."""
     disable_a2a: bool
     """Optional. If `True`, /a2a routes are removed, disabling default support to expose the deployment as an agent-to-agent (A2A) server.
 

libs/cli/schemas/schema.json

@@ -1065,6 +1065,10 @@
           ],
           "description": "Optional. Defines which headers are excluded from logging."
         },
+        "mcp": {
+          "$ref": "#/$defs/McpConfig",
+          "description": "Optional. Configure the MCP Tasks extension for elicitation over protocol 2026-07-28."
+        },
         "middleware_order": {
           "anyOf": [
             {
@@ -1170,6 +1174,42 @@
       },
       "required": []
     },
+    "McpConfig": {
+      "title": "McpConfig",
+      "type": "object",
+      "properties": {
+        "tasks": {
+          "$ref": "#/$defs/McpTasksConfig",
+          "description": "Polling, retention and URL settings for the MCP Tasks extension."
+        }
+      },
+      "required": [],
+      "description": "dict() -> new empty dictionary\ndict(mapping) -> new dictionary initialized from a mapping object's\n    (key, value) pairs\ndict(iterable) -> new dictionary initialized as if via:\n    d = {}\n    for k, v in iterable:\n        d[k] = v\ndict(**kwargs) -> new dictionary initialized with the name=value pairs\n    in the keyword argument list.  For example:  dict(one=1, two=2)"
+    },
+    "McpTasksConfig": {
+      "title": "McpTasksConfig",
+      "description": "Settings for authenticated MCP elicitation tasks.",
+      "type": "object",
+      "properties": {
+        "allow_insecure_urls": {
+          "type": "boolean",
+          "description": "Allow HTTP elicitation URLs for development. Defaults to False; production URLs require HTTPS."
+        },
+        "completed_grace_seconds": {
+          "type": "integer",
+          "description": "Result retention after completion, in seconds. Default 300; range 1\u201386400."
+        },
+        "poll_interval_ms": {
+          "type": "integer",
+          "description": "Suggested client polling interval, in milliseconds. Default 1000; range 100\u201360000."
+        },
+        "ttl_minutes": {
+          "type": "integer",
+          "description": "Maximum task lifetime, in minutes. Default 1440; range 1\u201310080."
+        }
+      },
+      "required": []
+    },
     "UvSource": {
       "title": "UvSource",
       "description": "Deployment source rooted at a uv project or workspace.",

libs/cli/schemas/schema.v0.json

@@ -1065,6 +1065,10 @@
           ],
           "description": "Optional. Defines which headers are excluded from logging."
         },
+        "mcp": {
+          "$ref": "#/$defs/McpConfig",
+          "description": "Optional. Configure the MCP Tasks extension for elicitation over protocol 2026-07-28."
+        },
         "middleware_order": {
           "anyOf": [
             {
@@ -1170,6 +1174,42 @@
       },
       "required": []
     },
+    "McpConfig": {
+      "title": "McpConfig",
+      "type": "object",
+      "properties": {
+        "tasks": {
+          "$ref": "#/$defs/McpTasksConfig",
+          "description": "Polling, retention and URL settings for the MCP Tasks extension."
+        }
+      },
+      "required": [],
+      "description": "dict() -> new empty dictionary\ndict(mapping) -> new dictionary initialized from a mapping object's\n    (key, value) pairs\ndict(iterable) -> new dictionary initialized as if via:\n    d = {}\n    for k, v in iterable:\n        d[k] = v\ndict(**kwargs) -> new dictionary initialized with the name=value pairs\n    in the keyword argument list.  For example:  dict(one=1, two=2)"
+    },
+    "McpTasksConfig": {
+      "title": "McpTasksConfig",
+      "description": "Settings for authenticated MCP elicitation tasks.",
+      "type": "object",
+      "properties": {
+        "allow_insecure_urls": {
+          "type": "boolean",
+          "description": "Allow HTTP elicitation URLs for development. Defaults to False; production URLs require HTTPS."
+        },
+        "completed_grace_seconds": {
+          "type": "integer",
+          "description": "Result retention after completion, in seconds. Default 300; range 1\u201386400."
+        },
+        "poll_interval_ms": {
+          "type": "integer",
+          "description": "Suggested client polling interval, in milliseconds. Default 1000; range 100\u201360000."
+        },
+        "ttl_minutes": {
+          "type": "integer",
+          "description": "Maximum task lifetime, in minutes. Default 1440; range 1\u201310080."
+        }
+      },
+      "required": []
+    },
     "UvSource": {
       "title": "UvSource",
       "description": "Deployment source rooted at a uv project or workspace.",

libs/langgraph/langgraph/types.py

@@ -14,14 +14,12 @@ from typing import (
     NamedTuple,
     TypeVar,
     final,
-    overload,
 )
 from warnings import warn
 
 from langchain_core.messages import AnyMessage
 from langchain_core.runnables import Runnable, RunnableConfig
 from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointMetadata
-from pydantic import TypeAdapter
 from typing_extensions import NotRequired, TypeAliasType, TypedDict, Unpack, deprecated
 from xxhash import xxh3_128_hexdigest
 
@@ -38,7 +36,6 @@ from langgraph.warnings import LangGraphDeprecatedSinceV10, LangGraphDeprecatedS
 # when used in standalone type aliases.
 StateT = TypeVar("StateT")
 OutputT = TypeVar("OutputT")
-ResponseT = TypeVar("ResponseT")
 
 if TYPE_CHECKING:
     from langgraph.pregel.protocol import PregelProtocol
@@ -599,19 +596,13 @@ class Interrupt:
     id: str
     """The ID of the interrupt. Can be used to resume the interrupt directly."""
 
-    response_schema: dict[str, Any] | None = None
-    """JSON Schema for the value expected when resuming this interrupt, if the graph provided one."""
-
     def __init__(
         self,
         value: Any,
         id: str = _DEFAULT_INTERRUPT_ID,
-        *,
-        response_schema: dict[str, Any] | None = None,
         **deprecated_kwargs: Unpack[DeprecatedKwargs],
     ) -> None:
         self.value = value
-        self.response_schema = response_schema
 
         if (
             (ns := deprecated_kwargs.get("ns", MISSING)) is not MISSING
@@ -623,14 +614,8 @@ class Interrupt:
             self.id = id
 
     @classmethod
-    def from_ns(
-        cls, value: Any, ns: str, *, response_schema: dict[str, Any] | None = None
-    ) -> Interrupt:
-        return cls(
-            value=value,
-            id=xxh3_128_hexdigest(ns.encode()),
-            response_schema=response_schema,
-        )
+    def from_ns(cls, value: Any, ns: str) -> Interrupt:
+        return cls(value=value, id=xxh3_128_hexdigest(ns.encode()))
 
     @property
     @deprecated("`interrupt_id` is deprecated. Use `id` instead.", category=None)
@@ -863,17 +848,7 @@ class Command(Generic[N], ToolOutputMixin):
     PARENT: ClassVar[Literal["__parent__"]] = "__parent__"
 
 
-@overload
-def interrupt(value: Any, *, response_schema: type[ResponseT]) -> ResponseT: ...
-
-
-@overload
-def interrupt(value: Any, *, response_schema: dict[str, Any] | None = None) -> Any: ...
-
-
-def interrupt(
-    value: Any, *, response_schema: dict[str, Any] | type[Any] | None = None
-) -> Any:
+def interrupt(value: Any) -> Any:
     """Interrupt the graph with a resumable exception from within a node.
 
     The `interrupt` function enables human-in-the-loop workflows by pausing graph
@@ -943,7 +918,7 @@ def interrupt(
         for chunk in graph.stream({\"foo\": \"abc\"}, config):
             print(chunk)
 
-        # > {'__interrupt__': (Interrupt(value='what is your age?', id='45fda8478b2ef754419799e10992af06', response_schema=None),)}
+        # > {'__interrupt__': (Interrupt(value='what is your age?', id='45fda8478b2ef754419799e10992af06'),)}
 
         command = Command(resume=\"some input from a human!!!\")
 
@@ -956,20 +931,12 @@ def interrupt(
 
     Args:
         value: The value to surface to the client when the graph is interrupted.
-        response_schema: Optional schema for the value expected on resume, surfaced
-            to clients so they can render a typed input form. Accepts a JSON Schema
-            `dict` (used as-is, resume values are not validated), or a Pydantic model
-            class, `TypedDict`, or dataclass, which are converted to JSON Schema for
-            clients and used to validate the resume value; the validated object is
-            what `interrupt` returns.
 
     Returns:
-        Any: On subsequent invocations within the same node (same task to be precise), returns the value provided during the first invocation,
-            validated against `response_schema` when one that supports validation was given.
+        Any: On subsequent invocations within the same node (same task to be precise), returns the value provided during the first invocation
 
     Raises:
         GraphInterrupt: On the first invocation within the node, halts execution and surfaces the provided value to the client.
-        pydantic.ValidationError: When a resume value does not match a Pydantic model, `TypedDict`, or dataclass `response_schema`.
     """
     from langgraph._internal._constants import (
         CONFIG_KEY_CHECKPOINT_NS,
@@ -981,36 +948,27 @@ def interrupt(
     from langgraph.errors import GraphInterrupt
 
     conf = get_config()["configurable"]
-    adapter = (
-        None
-        if response_schema is None or isinstance(response_schema, dict)
-        else TypeAdapter(response_schema)
-    )
     # track interrupt index
     scratchpad = conf[CONFIG_KEY_SCRATCHPAD]
     idx = scratchpad.interrupt_counter()
     # find previous resume values
     if scratchpad.resume:
         if idx < len(scratchpad.resume):
-            v = scratchpad.resume[idx]
-            validated = adapter.validate_python(v) if adapter else v
             conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)])
-            return validated
+            return scratchpad.resume[idx]
     # find current resume value
     v = scratchpad.get_null_resume(True)
     if v is not None:
         assert len(scratchpad.resume) == idx, (scratchpad.resume, idx)
-        validated = adapter.validate_python(v) if adapter else v
         scratchpad.resume.append(v)
         conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)])
-        return validated
+        return v
     # no resume value found
     raise GraphInterrupt(
         (
             Interrupt.from_ns(
                 value=value,
                 ns=conf[CONFIG_KEY_CHECKPOINT_NS],
-                response_schema=adapter.json_schema() if adapter else response_schema,
             ),
         )
     )

libs/langgraph/tests/test_interruption.py

@@ -1,14 +1,9 @@
-from dataclasses import dataclass
-from typing import Any
-
 import pytest
 from langgraph.checkpoint.base import BaseCheckpointSaver
-from pydantic import BaseModel, ValidationError
 from typing_extensions import TypedDict
 
 from langgraph.graph import END, START, StateGraph
-from langgraph.types import Command, Durability, Interrupt, interrupt
-from tests.any_str import AnyStr
+from langgraph.types import Durability
 
 pytestmark = pytest.mark.anyio
 
@@ -95,116 +90,3 @@ async def test_interruption_without_state_updates_async(
     assert (await graph.aget_state(thread)).next == ()
     n_checkpoints = len([c async for c in graph.aget_state_history(thread)])
     assert n_checkpoints == (5 if durability != "exit" else 3)
-
-
-class Decision(BaseModel):
-    approved: bool
-    note: str | None = None
-
-
-class DecisionDict(TypedDict):
-    approved: bool
-
-
-@dataclass
-class DecisionData:
-    approved: bool
-
-
-RAW_SCHEMA = {"type": "object", "properties": {"approved": {"type": "boolean"}}}
-
-
-@pytest.mark.parametrize(
-    ("response_schema", "expected_schema", "expected_answer"),
-    [
-        (None, None, {"approved": True, "extra": 1}),
-        (RAW_SCHEMA, RAW_SCHEMA, {"approved": True, "extra": 1}),
-        (Decision, Decision.model_json_schema(), Decision(approved=True)),
-        (
-            DecisionDict,
-            {
-                "properties": {"approved": {"title": "Approved", "type": "boolean"}},
-                "required": ["approved"],
-                "title": "DecisionDict",
-                "type": "object",
-            },
-            {"approved": True},
-        ),
-        (
-            DecisionData,
-            {
-                "properties": {"approved": {"title": "Approved", "type": "boolean"}},
-                "required": ["approved"],
-                "title": "DecisionData",
-                "type": "object",
-            },
-            DecisionData(approved=True),
-        ),
-    ],
-    ids=["none", "raw_dict", "pydantic", "typeddict", "dataclass"],
-)
-def test_interrupt_response_schema(
-    sync_checkpointer: BaseCheckpointSaver,
-    response_schema: Any,
-    expected_schema: dict[str, Any] | None,
-    expected_answer: Any,
-) -> None:
-    class State(TypedDict):
-        answer: Any
-
-    def node(state: State) -> State:
-        return {
-            "answer": interrupt(
-                {"question": "approve?"}, response_schema=response_schema
-            )
-        }
-
-    graph = (
-        StateGraph(State)
-        .add_node("node", node)
-        .add_edge(START, "node")
-        .compile(checkpointer=sync_checkpointer)
-    )
-    config = {"configurable": {"thread_id": "1"}}
-    expected = Interrupt(
-        value={"question": "approve?"}, id=AnyStr(), response_schema=expected_schema
-    )
-
-    assert list(graph.stream({"answer": None}, config)) == [
-        {"__interrupt__": (expected,)}
-    ]
-    assert graph.get_state(config).tasks[0].interrupts == (expected,)
-    assert graph.invoke(Command(resume={"approved": True, "extra": 1}), config) == {
-        "answer": expected_answer
-    }
-
-
-@pytest.mark.parametrize("resume_style", ["null", "map"])
-def test_interrupt_response_schema_rejects_invalid_resume(
-    sync_checkpointer: BaseCheckpointSaver, resume_style: str
-) -> None:
-    class State(TypedDict):
-        answer: Any
-
-    def node(state: State) -> State:
-        return {"answer": interrupt("approve?", response_schema=Decision)}
-
-    graph = (
-        StateGraph(State)
-        .add_node("node", node)
-        .add_edge(START, "node")
-        .compile(checkpointer=sync_checkpointer)
-    )
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke({"answer": None}, config)
-    [pending] = graph.get_state(config).tasks[0].interrupts
-
-    def resume(value: dict[str, Any]) -> Command:
-        return Command(resume=value if resume_style == "null" else {pending.id: value})
-
-    with pytest.raises(ValidationError, match="approved"):
-        graph.invoke(resume({"approved": "nope"}), config)
-
-    assert graph.invoke(resume({"approved": False}), config) == {
-        "answer": Decision(approved=False)
-    }

libs/langgraph/tests/test_pregel.py

@@ -5583,7 +5583,6 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver):
                 "interrupts": [
                     {
                         "id": AnyStr(),
-                        "response_schema": None,
                         "value": "test",
                     },
                 ],
@@ -5628,7 +5627,6 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver):
                         "interrupts": (
                             {
                                 "id": AnyStr(),
-                                "response_schema": None,
                                 "value": "test",
                             },
                         ),

libs/sdk-py/langgraph_sdk/schema.py

@@ -295,8 +295,6 @@ class Interrupt(TypedDict):
     """The value associated with the interrupt."""
     id: str
     """The ID of the interrupt. Can be used to resume the interrupt."""
-    response_schema: NotRequired[dict[str, Any]]
-    """JSON Schema for the value expected when resuming this interrupt, if the graph provided one."""
 
 
 class Thread(TypedDict):