ag-ui-protocol · Sri01728 · Jul 14, 2026 · Jul 14, 2026 · Jul 14, 2026 · Jul 14, 2026
@@ -31,17 +31,19 @@ add_claude_fastapi_endpoint(app=app, adapter=adapter, path="/my_agent")
 - **Event cleanup** - Hanging events (tool calls, reasoning blocks) automatically closed on stream end
 - **Custom tools via MCP** - Define custom tools using Claude SDK's @tool decorator
 - **Forwarded props** - Per-run option overrides with security whitelist
+- **Interrupt/resume (AG-UI contract)** - Opt in with `emit_interrupt_outcome=True`. A tool deferred by a PreToolUse hook (`permissionDecision: "defer"`) is surfaced as a `RunFinishedInterruptOutcome`, and `RunAgentInput.resume[]` verdicts are honoured on the resuming run. Default-off for back-compat.
 
 ## Examples
 
-The integration includes 5 example agents:
+The integration includes 6 example agents:
 
 | Route | Description | Features |
 |-------|-------------|----------|
 | `/agentic_chat` | Basic conversational assistant | Simple chat |
 | `/backend_tool_rendering` | Weather tool (backend MCP) | Backend tool execution, tool rendering |
 | `/shared_state` | Recipe collaboration | Bidirectional state sync, ag_ui_update_state |
 | `/human_in_the_loop` | Task planning with approval | Frontend tools, step tracking, approval workflow |
+| `/interrupt` | Destructive tool behind approval | `defer` -> interrupt outcome, `resume[]` verdict, frozen args |
 | `/tool_based_generative_ui` | Frontend tool rendering | Dynamic frontend tools, generative UI |
 
 ## Running the Examples

@@ -18,6 +18,12 @@
 
 from .adapter import ClaudeAgentAdapter
 from .endpoint import add_claude_fastapi_endpoint
+from .interrupts import (
+    deferred_tool_use_to_interrupt,
+    interrupt_id_for_tool_use,
+    tool_use_id_from_interrupt_id,
+    INTERRUPT_REASON_TOOL_CALL,
+)
 from .config import (
     ALLOWED_FORWARDED_PROPS,
     STATE_MANAGEMENT_TOOL_NAME,
@@ -31,6 +37,11 @@
 __all__ = [
     "ClaudeAgentAdapter",
     "add_claude_fastapi_endpoint",
+    # Interrupt/resume bridge (AG-UI interrupt contract)
+    "deferred_tool_use_to_interrupt",
+    "interrupt_id_for_tool_use",
+    "tool_use_id_from_interrupt_id",
+    "INTERRUPT_REASON_TOOL_CALL",
     # Configuration constants
     "ALLOWED_FORWARDED_PROPS",
     "STATE_MANAGEMENT_TOOL_NAME",

@@ -33,6 +33,13 @@
     ReasoningMessageEndEvent,
     ReasoningEndEvent,
     ReasoningEncryptedValueEvent,
+    RunFinishedInterruptOutcome,
+)
+
+from .interrupts import (
+    deferred_tool_use_to_interrupt,
+    tool_use_id_from_interrupt_id,
+    is_resume_resolved,
 )
 
 if TYPE_CHECKING:
@@ -87,13 +94,31 @@ def __init__(
         max_workers: int = 1000,
         worker_ttl_seconds: float = 1800,   # 30 min
         query_timeout_seconds: Optional[float] = 300,   # 5 min; bounds a hung/slow worker
+        emit_interrupt_outcome: bool = False,
     ):
         self.name = name
         self.description = description
         self._options = options
         self._max_workers = max_workers
         self._worker_ttl_seconds = worker_ttl_seconds
         self._query_timeout_seconds = query_timeout_seconds
+        # When True, a tool deferred by a PreToolUse hook is surfaced as an
+        # AG-UI ``RunFinishedInterruptOutcome`` and the adapter honours
+        # ``RunAgentInput.resume[]``. Default-off for back-compat: a client that
+        # predates the interrupt contract must not be handed an ``outcome`` it
+        # cannot resume (it would strand the run). Mirrors LangGraph's
+        # ``emit_interrupt_outcome`` opt-in.
+        self._emit_interrupt_outcome = emit_interrupt_outcome
+        # thread_id -> {deferred_tool_use_id: verdict dict}. Populated from
+        # RunAgentInput.resume[] at the start of a resuming run; read by the
+        # caller-registered PreToolUse hook (via ``resume_verdict_for``) to
+        # allow or deny the re-fired deferred call.
+        self._resume_verdicts: Dict[str, Dict[str, Dict[str, Any]]] = {}
+        # thread_id -> last known Claude SDK session_id, captured from the worker
+        # after each run. On an interrupt-resume run this is passed to
+        # build_options(resume_session_id=...) so a fresh worker resumes the
+        # persisted session and re-fires the frozen deferred call.
+        self._claude_session_ids: Dict[str, str] = {}
         # thread_id -> {"worker": SessionWorker, "last_used": datetime, "active": bool, "active_runs": int}
         self._workers: Dict[str, Dict] = {}
         self._state_locks: Dict[str, asyncio.Lock] = {}
@@ -221,6 +246,66 @@ async def clear_session(self, thread_id: str) -> None:
         self._per_thread_state.pop(thread_id, None)
         self._drop_thread_results(thread_id)
 
+    def _build_interrupt_outcome(
+        self,
+        run_result: Optional[Dict[str, Any]],
+        input_data: RunAgentInput,
+    ) -> Optional[RunFinishedInterruptOutcome]:
+        """Build the RUN_FINISHED interrupt outcome for a deferred tool call.
+
+        Returns ``None`` (so RUN_FINISHED stays a plain success) unless the
+        interrupt contract is opted in AND this run halted on a deferred tool.
+        The proposed tool call has already been streamed as
+        TOOL_CALL_START/ARGS/END, so the client sees the frozen args before
+        the interrupt arrives.
+        """
+        if not self._emit_interrupt_outcome:
+            return None
+        if not run_result:
+            return None
+        deferred = run_result.get("deferred_tool_use")
+        if deferred is None:
+            return None
+        interrupt = deferred_tool_use_to_interrupt(deferred, tools=input_data.tools)
+        return RunFinishedInterruptOutcome(type="interrupt", interrupts=[interrupt])
+
+    def _ingest_resume(self, thread_id: str, input_data: RunAgentInput) -> None:
+        """Record ``RunAgentInput.resume[]`` verdicts for this thread.
+
+        Each ``ResumeEntry`` is keyed by the deferred tool-use id recovered from
+        its ``interrupt_id`` so the re-fired PreToolUse hook can look it up.
+        A missing/empty resume array clears any stale verdicts for the thread.
+        """
+        resume = getattr(input_data, "resume", None)
+        if not resume:
+            self._resume_verdicts.pop(thread_id, None)
+            return
+        verdicts: Dict[str, Dict[str, Any]] = {}
+        for entry in resume:
+            interrupt_id = getattr(entry, "interrupt_id", None)
+            if not interrupt_id:
+                continue
+            tool_use_id = tool_use_id_from_interrupt_id(interrupt_id)
+            verdicts[tool_use_id] = {
+                "status": getattr(entry, "status", None),
+                "payload": getattr(entry, "payload", None),
+                "resolved": is_resume_resolved(getattr(entry, "status", "")),
+            }
+        self._resume_verdicts[thread_id] = verdicts
+
+    def resume_verdict_for(
+        self,
+        thread_id: str,
+        tool_use_id: str,
+    ) -> Optional[Dict[str, Any]]:
+        """Return the resume verdict for a deferred tool-use, or ``None``.
+
+        A caller-registered PreToolUse hook calls this when a deferred tool
+        re-fires on resume: a ``resolved`` verdict means allow the frozen call,
+        anything else (``cancelled``) means deny it.
+        """
+        return self._resume_verdicts.get(thread_id, {}).get(tool_use_id)
+
     async def run(self, input_data: RunAgentInput) -> AsyncIterator[BaseEvent]:
         """Run the agent and yield AG-UI events."""
         from .utils import process_messages
@@ -229,6 +314,26 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[BaseEvent]:
         run_id = input_data.run_id or str(uuid.uuid4())
         result_key = (thread_id, run_id)
 
+        # ── Resume handling (AG-UI interrupt contract) ──
+        # A resume request continues the persisted Claude session (already keyed
+        # by thread_id) so the frozen deferred tool re-fires PreToolUse. The
+        # per-interrupt verdicts are published on a per-thread map keyed by the
+        # deferred tool-use id, which the caller-registered PreToolUse hook reads
+        # to allow (with the frozen args) or deny the re-fired call. Args are
+        # NEVER re-derived from the resumed model turn — they stay bound to the
+        # frozen DeferredToolUse.input, so resume cannot open an args-swap window.
+        if self._emit_interrupt_outcome:
+            self._ingest_resume(thread_id, input_data)
+        # A resume run must NOT reuse the live in-process worker: the SDK only
+        # re-fires a deferred tool via ``ClaudeAgentOptions.resume`` at connect
+        # time, so we force a fresh worker seeded with the persisted Claude
+        # session_id below.
+        is_resume_run = bool(
+            self._emit_interrupt_outcome
+            and getattr(input_data, "resume", None)
+            and self._claude_session_ids.get(thread_id)
+        )
+
         # ── Run-admission serialization (Fix 1) ──
         # Acquire the per-thread RUN lock at admission — BEFORE worker.query() /
         # RUN_STARTED — and hold it across the WHOLE run, releasing in the
@@ -276,6 +381,31 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[BaseEvent]:
             # reusing it would hang forever on a queue nothing drains. Evict the
             # dead worker and fall through to creating a fresh one.
             entry = self._workers.get(thread_id)
+            # Interrupt-resume: never reuse the live worker. The SDK only
+            # re-fires a deferred tool via ClaudeAgentOptions.resume at connect
+            # time, so evict any live worker and fall through to creating a
+            # fresh one seeded with the persisted Claude session_id.
+            if is_resume_run and entry is not None:
+                if entry.get("active_runs", 0) > 0:
+                    logger.error(
+                        f"Cannot resume thread={thread_id}: a peer run is still active"
+                    )
+                    yield RunErrorEvent(
+                        type=EventType.RUN_ERROR,
+                        thread_id=thread_id,
+                        run_id=run_id,
+                        message=(
+                            f"cannot resume run on thread {thread_id}: its worker "
+                            f"has another run still active"
+                        ),
+                    )
+                    return
+                logger.debug(f"Evicting live worker for thread={thread_id} to resume persisted session")
+                stale = self._workers.pop(thread_id, None)
+                if stale is not None:
+                    await stale["worker"].stop()
+                self._state_locks.pop(thread_id, None)
+                entry = None
             if entry is not None and not entry["worker"].is_alive():
                 if entry.get("active_runs", 0) > 0:
                     # DEFENSE-IN-DEPTH / UNREACHABLE under run-admission
@@ -323,7 +453,8 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[BaseEvent]:
                     entry = None
 
             if entry is None:
-                options = self.build_options(input_data, thread_id=thread_id)
+                resume_session_id = self._claude_session_ids.get(thread_id) if is_resume_run else None
+                options = self.build_options(input_data, thread_id=thread_id, resume_session_id=resume_session_id)
                 worker = SessionWorker(thread_id, options)
                 await worker.start()
                 # ``active_runs`` is a refcount of in-flight run() invocations
@@ -399,11 +530,20 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[BaseEvent]:
 
             # Emit RUN_FINISHED — read THIS run's own result (keyed per-run, so a
             # serialized peer on the same thread cannot have clobbered it). (Fix 4)
+            run_result = self._per_run_result.get(result_key, None)
+            outcome = self._build_interrupt_outcome(run_result, input_data)
+            # Remember the Claude SDK session_id so a subsequent interrupt-resume
+            # run can seed build_options(resume_session_id=...) and re-fire the
+            # frozen deferred call. Only meaningful when the interrupt contract
+            # is enabled; harmless otherwise.
+            if self._emit_interrupt_outcome and getattr(worker, "session_id", None):
+                self._claude_session_ids[thread_id] = worker.session_id
             yield RunFinishedEvent(
                 type=EventType.RUN_FINISHED,
                 thread_id=thread_id,
                 run_id=run_id,
-                result=self._per_run_result.get(result_key, None),
+                result=run_result,
+                outcome=outcome,
             )
 
         except asyncio.TimeoutError as e:
@@ -474,8 +614,16 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[BaseEvent]:
             # run can proceed. We acquired it unconditionally before this try.
             run_lock.release()
 
-    def build_options(self, input_data: Optional[RunAgentInput] = None, thread_id: Optional[str] = None) -> "ClaudeAgentOptions":
-        """Build ClaudeAgentOptions from base config + RunAgentInput."""
+    def build_options(self, input_data: Optional[RunAgentInput] = None, thread_id: Optional[str] = None, resume_session_id: Optional[str] = None) -> "ClaudeAgentOptions":
+        """Build ClaudeAgentOptions from base config + RunAgentInput.
+
+        When ``resume_session_id`` is set (a resuming run under the interrupt
+        contract), it is passed to the SDK as ``resume`` so ``connect()``
+        materializes the persisted Claude session and replays the transcript —
+        which re-fires the frozen deferred tool call through PreToolUse. This is
+        the ONLY way to re-drive a deferred call; a live client's ``query()``
+        continues the conversation but does not resume a deferred tool.
+        """
         from claude_agent_sdk import ClaudeAgentOptions, create_sdk_mcp_server
 
         # Start with sensible defaults
@@ -540,6 +688,12 @@ def build_options(self, input_data: Optional[RunAgentInput] = None, thread_id: O
         # Remove api_key from options kwargs (handled via environment variable)
         merged_kwargs.pop("api_key", None)
         logger.debug(f"Merged kwargs after pop: {merged_kwargs}")
+
+        # Resume the persisted Claude session for an interrupt-resume run so the
+        # frozen deferred tool call re-fires PreToolUse (see method docstring).
+        if resume_session_id:
+            merged_kwargs["resume"] = resume_session_id
+            logger.debug(f"Resuming Claude session {resume_session_id} for deferred-tool replay")
 
         # Apply forwarded_props as per-run overrides (before adding dynamic tools)
         if input_data and input_data.forwarded_props:
@@ -1054,6 +1208,10 @@ def flush_pending_msg():
                     "total_cost_usd": getattr(message, 'total_cost_usd', None),
                     "usage": getattr(message, 'usage', None),
                     "structured_output": getattr(message, 'structured_output', None),
+                    # DeferredToolUse | None. When present the run halted on a
+                    # PreToolUse ``defer``; run() turns this into an AG-UI
+                    # interrupt outcome (only if emit_interrupt_outcome is set).
+                    "deferred_tool_use": getattr(message, 'deferred_tool_use', None),
                 }
 
                 if not has_streamed_text and result_text:

@@ -0,0 +1,106 @@
+"""Bridge Claude Agent SDK ``defer`` into the AG-UI interrupt/resume contract.
+
+The Claude Agent SDK has no in-flight tool suspension. When a ``PreToolUse``
+hook returns ``permissionDecision: "defer"``, the run halts at the tool
+boundary and the terminating ``ResultMessage`` carries a ``DeferredToolUse``
+(the frozen ``id``/``name``/``input`` of the proposed call) plus a
+``session_id``. Resuming means starting a fresh ``query(resume=session_id)``:
+the same frozen tool call fires ``PreToolUse`` again, and the hook then allows
+or denies it based on the resume verdict.
+
+This module is the thin translation layer between that primitive and the
+AG-UI wire types, matching how ``ag_ui_langgraph.interrupts`` reshapes
+LangGraph's native ``Interrupt`` into ``ag_ui.core.Interrupt``. It performs no
+suspension itself — it only reshapes a halt into the standard contract.
+"""
+
+from typing import Any, Dict, List, Optional
+
+from ag_ui.core import Interrupt
+
+# Prefix used to derive the AG-UI interrupt id from the deferred tool-use id.
+# The interrupt id (the approval-request id) is intentionally distinct from the
+# tool_call_id (the proposed call id) — the client resolves an interrupt by its
+# own id, not by the tool call id. This mirrors the AG-UI reference contract.
+_INTERRUPT_ID_PREFIX = "interrupt_"
+
+# ``reason`` value for interrupts raised by a deferred tool call.
+INTERRUPT_REASON_TOOL_CALL = "tool_call"
+
+
+def interrupt_id_for_tool_use(deferred_tool_use_id: str) -> str:
+    """Return the AG-UI interrupt id for a deferred tool-use id."""
+    return f"{_INTERRUPT_ID_PREFIX}{deferred_tool_use_id}"
+
+
+def tool_use_id_from_interrupt_id(interrupt_id: str) -> str:
+    """Recover the deferred tool-use id from an AG-UI interrupt id.
+
+    Inverse of :func:`interrupt_id_for_tool_use`. If the id does not carry the
+    expected prefix (e.g. a client sent a bare tool_call_id), it is returned
+    unchanged so resume stays best-effort rather than raising.
+    """
+    if interrupt_id.startswith(_INTERRUPT_ID_PREFIX):
+        return interrupt_id[len(_INTERRUPT_ID_PREFIX):]
+    return interrupt_id
+
+
+def _response_schema_for_tool(
+    tool_name: str,
+    tools: Optional[List[Any]],
+) -> Optional[Dict[str, Any]]:
+    """Best-effort JSON Schema describing the resume payload for ``tool_name``.
+
+    Pulls the tool's declared input schema from the run's ``tools`` so a client
+    can render an approval form. Returns ``None`` when the tool or its schema
+    cannot be found; a missing schema must never fail the run.
+    """
+    if not tools:
+        return None
+    for tool in tools:
+        name = getattr(tool, "name", None)
+        if name is None and isinstance(tool, dict):
+            name = tool.get("name")
+        if name != tool_name:
+            continue
+        schema = getattr(tool, "parameters", None)
+        if schema is None and isinstance(tool, dict):
+            schema = tool.get("parameters")
+        if isinstance(schema, dict):
+            return schema
+        return None
+    return None
+
+
+def deferred_tool_use_to_interrupt(
+    deferred_tool_use: Any,
+    tools: Optional[List[Any]] = None,
+    message: Optional[str] = None,
+) -> Interrupt:
+    """Convert a Claude SDK ``DeferredToolUse`` into an AG-UI ``Interrupt``.
+
+    Args:
+        deferred_tool_use: The ``DeferredToolUse`` from the terminating
+            ``ResultMessage`` (carries ``id``, ``name``, ``input``).
+        tools: The run's declared tools, used to derive ``response_schema``.
+        message: Optional human-readable prompt for the approval surface.
+
+    Returns:
+        An ``Interrupt`` whose ``tool_call_id`` is the frozen deferred call id
+        and whose ``id`` is the distinct approval-request id.
+    """
+    tool_use_id = deferred_tool_use.id
+    tool_name = deferred_tool_use.name
+    return Interrupt(
+        id=interrupt_id_for_tool_use(tool_use_id),
+        reason=INTERRUPT_REASON_TOOL_CALL,
+        message=message,
+        tool_call_id=tool_use_id,
+        response_schema=_response_schema_for_tool(tool_name, tools),
+        metadata={"tool_name": tool_name},
+    )
+
+
+def is_resume_resolved(status: str) -> bool:
+    """Return True when a ``ResumeEntry.status`` means "proceed"."""
+    return status == "resolved"