diff --git a/integrations/claude-agent-sdk/python/README.md b/integrations/claude-agent-sdk/python/README.md index 7a5f5c54b4..3b9c417d73 100644 --- a/integrations/claude-agent-sdk/python/README.md +++ b/integrations/claude-agent-sdk/python/README.md @@ -31,10 +31,11 @@ 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 | |-------|-------------|----------| @@ -42,6 +43,7 @@ The integration includes 5 example agents: | `/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 diff --git a/integrations/claude-agent-sdk/python/ag_ui_claude_sdk/__init__.py b/integrations/claude-agent-sdk/python/ag_ui_claude_sdk/__init__.py index 8b5c4d278d..5e9ff1af13 100644 --- a/integrations/claude-agent-sdk/python/ag_ui_claude_sdk/__init__.py +++ b/integrations/claude-agent-sdk/python/ag_ui_claude_sdk/__init__.py @@ -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", diff --git a/integrations/claude-agent-sdk/python/ag_ui_claude_sdk/adapter.py b/integrations/claude-agent-sdk/python/ag_ui_claude_sdk/adapter.py index e38a3a91c8..8a368538cd 100644 --- a/integrations/claude-agent-sdk/python/ag_ui_claude_sdk/adapter.py +++ b/integrations/claude-agent-sdk/python/ag_ui_claude_sdk/adapter.py @@ -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,6 +94,7 @@ 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 @@ -94,6 +102,23 @@ def __init__( 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 @@ -419,14 +550,20 @@ async def run(self, input_data: RunAgentInput) -> AsyncIterator[BaseEvent]: code=str(api_error_status) if api_error_status is not None else None, ) else: - # 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) + 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: @@ -497,8 +634,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 @@ -563,6 +708,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: @@ -1080,6 +1231,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), # Best-effort: absent from ResultMessage on claude-agent-sdk # versions predating the field (including the current pin # floor), in which case RUN_ERROR.code is None. diff --git a/integrations/claude-agent-sdk/python/ag_ui_claude_sdk/interrupts.py b/integrations/claude-agent-sdk/python/ag_ui_claude_sdk/interrupts.py new file mode 100644 index 0000000000..43481a14ce --- /dev/null +++ b/integrations/claude-agent-sdk/python/ag_ui_claude_sdk/interrupts.py @@ -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" diff --git a/integrations/claude-agent-sdk/python/examples/agents/__init__.py b/integrations/claude-agent-sdk/python/examples/agents/__init__.py index b40a45cc86..d108a3c2b6 100644 --- a/integrations/claude-agent-sdk/python/examples/agents/__init__.py +++ b/integrations/claude-agent-sdk/python/examples/agents/__init__.py @@ -9,6 +9,7 @@ from .backend_tool_rendering import create_backend_tool_adapter from .shared_state import create_shared_state_adapter from .human_in_the_loop import create_human_in_the_loop_adapter +from .interrupt import create_interrupt_adapter from .tool_based_generative_ui import create_tool_based_generative_ui_adapter __all__ = [ @@ -16,5 +17,6 @@ "create_backend_tool_adapter", "create_shared_state_adapter", "create_human_in_the_loop_adapter", + "create_interrupt_adapter", "create_tool_based_generative_ui_adapter", ] diff --git a/integrations/claude-agent-sdk/python/examples/agents/interrupt.py b/integrations/claude-agent-sdk/python/examples/agents/interrupt.py new file mode 100644 index 0000000000..1c8a967120 --- /dev/null +++ b/integrations/claude-agent-sdk/python/examples/agents/interrupt.py @@ -0,0 +1,102 @@ +"""Interrupt / resume agent configuration (AG-UI interrupt contract). + +Demonstrates the ``defer``-based interrupt bridge: + +1. The model calls the backend ``delete_file`` tool. +2. A PreToolUse hook returns ``permissionDecision: "defer"`` the first time it + sees a given tool call, so the run halts at the tool boundary. The adapter + (constructed with ``emit_interrupt_outcome=True``) turns the resulting + ``DeferredToolUse`` into a ``RunFinishedInterruptOutcome`` — the frontend + receives ``outcome: {type: "interrupt", interrupts: [...]}`` with a + ``response_schema`` it can render as an approval form. +3. The user resolves the interrupt; the client sends the next request with + ``resume: [{interrupt_id, status: "resolved"|"cancelled", payload}]`` on the + same ``thread_id``. +4. The persisted Claude session continues; the SAME frozen tool call re-fires + the PreToolUse hook. The hook reads the adapter's resume verdict: + ``resolved`` -> ``allow`` (with the frozen args), otherwise ``deny``. + +Enforcement lives in the hook, not the prompt: the model cannot execute the +gated tool until a real resume verdict exists, and the executed args stay bound +to the frozen ``DeferredToolUse.input`` (resume carries only the verdict). +""" + +import json +from typing import Any + +from claude_agent_sdk import tool, create_sdk_mcp_server, HookMatcher +from ag_ui_claude_sdk import ClaudeAgentAdapter +from ag_ui_claude_sdk.interrupts import tool_use_id_from_interrupt_id # noqa: F401 (documents the id mapping) +from .constants import DEFAULT_DISALLOWED_TOOLS + +GATED_TOOL = "delete_file" + + +@tool(GATED_TOOL, "Delete a file at the given path", {"path": str}) +async def delete_file(args: dict[str, Any]) -> dict[str, Any]: + """Mock destructive tool — only runs after the interrupt is resolved.""" + path = args.get("path", "") + return {"content": [{"type": "text", "text": f"Deleted {path}"}], "path": path} + + +file_ops_server = create_sdk_mcp_server("file_ops", "1.0.0", tools=[delete_file]) + + +def create_interrupt_adapter() -> ClaudeAgentAdapter: + """Create adapter for the interrupt/resume demo.""" + adapter = ClaudeAgentAdapter( + name="interrupt", + description="Destructive tool gated behind an AG-UI interrupt", + emit_interrupt_outcome=True, + options={ + "model": "claude-haiku-4-5", + "system_prompt": ( + "You help with file operations. When asked to delete a file, " + f"call the {GATED_TOOL} tool with the path. The deletion is " + "gated behind human approval; just make the call and let the " + "approval flow handle the rest." + ), + "mcp_servers": {"file_ops": file_ops_server}, + "allowed_tools": [f"mcp__file_ops__{GATED_TOOL}"], + "disallowed_tools": list(DEFAULT_DISALLOWED_TOOLS), + }, + ) + + async def gate_hook(input_data: dict, tool_use_id: str, context: Any) -> dict: + """PreToolUse hook: defer the gated tool until a resume verdict exists. + + The adapter records resume verdicts per thread keyed by the deferred + tool-use id. On the resumed run the same tool_use_id re-fires here, so + we look it up and allow (frozen args) or deny. + """ + tool_name = input_data.get("tool_name", "") + if not tool_name.endswith(GATED_TOOL): + return {} + + thread_id = getattr(context, "session_id", None) or getattr(context, "thread_id", "") + verdict = adapter.resume_verdict_for(thread_id, tool_use_id) + + if verdict is None: + # First sighting — pause the run and surface the interrupt. + return {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "defer"}} + + if verdict["resolved"]: + # Approved: run the frozen call exactly as proposed. + return {"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "allow"}} + + # Cancelled: refuse the call; the model is told and can respond. + return { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "deny", + "permissionDecisionReason": "User declined the file deletion.", + } + } + + # Register the gate on the adapter's base options so every run carries it. + opts = adapter._options or {} + if isinstance(opts, dict): + hooks = opts.setdefault("hooks", {}) + hooks["PreToolUse"] = [HookMatcher(hooks=[gate_hook])] + + return adapter diff --git a/integrations/claude-agent-sdk/python/examples/server.py b/integrations/claude-agent-sdk/python/examples/server.py index e122ae89fb..8ceaca9ac6 100644 --- a/integrations/claude-agent-sdk/python/examples/server.py +++ b/integrations/claude-agent-sdk/python/examples/server.py @@ -16,6 +16,7 @@ from agents.backend_tool_rendering import create_backend_tool_adapter from agents.shared_state import create_shared_state_adapter from agents.human_in_the_loop import create_human_in_the_loop_adapter +from agents.interrupt import create_interrupt_adapter from agents.tool_based_generative_ui import create_tool_based_generative_ui_adapter app = FastAPI(title="Claude Agent SDK Server") @@ -40,6 +41,7 @@ "backend_tool_rendering": create_backend_tool_adapter(), "shared_state": create_shared_state_adapter(), "human_in_the_loop": create_human_in_the_loop_adapter(), + "interrupt": create_interrupt_adapter(), "tool_based_generative_ui": create_tool_based_generative_ui_adapter(), } diff --git a/integrations/claude-agent-sdk/python/tests/test_interrupts.py b/integrations/claude-agent-sdk/python/tests/test_interrupts.py new file mode 100644 index 0000000000..0a7339d96b --- /dev/null +++ b/integrations/claude-agent-sdk/python/tests/test_interrupts.py @@ -0,0 +1,155 @@ +"""Tests for the AG-UI interrupt/resume bridge in the Claude adapter. + +The bridge wraps the Claude SDK ``defer`` primitive: a deferred tool halts the +run, and the adapter surfaces it as an AG-UI ``RunFinishedInterruptOutcome``. +Resume verdicts arriving on ``RunAgentInput.resume[]`` are recorded so the +caller's re-fired PreToolUse hook can allow (frozen args) or deny the call. + +These tests exercise the pure translation + verdict-recording logic; no LLM or +SessionWorker is involved. +""" + +import pytest + +from ag_ui.core import ( + RunAgentInput, + ResumeEntry, + Interrupt, + RunFinishedInterruptOutcome, + Tool, +) +from claude_agent_sdk.types import DeferredToolUse + +from ag_ui_claude_sdk.adapter import ClaudeAgentAdapter +from ag_ui_claude_sdk.interrupts import ( + deferred_tool_use_to_interrupt, + interrupt_id_for_tool_use, + tool_use_id_from_interrupt_id, + INTERRUPT_REASON_TOOL_CALL, +) + + +def _deferred(tool_id="toolu_1", name="delete_file", args=None): + return DeferredToolUse(id=tool_id, name=name, input=args or {"path": "/etc/hosts"}) + + +def _resume_input(thread_id, entries, run_id="run-2"): + return RunAgentInput( + thread_id=thread_id, + run_id=run_id, + messages=[], + tools=[], + state=None, + context=[], + forwarded_props={}, + resume=entries, + ) + + +class TestInterruptIdRoundTrip: + def test_id_prefix_and_inverse(self): + iid = interrupt_id_for_tool_use("toolu_abc") + assert iid == "interrupt_toolu_abc" + assert tool_use_id_from_interrupt_id(iid) == "toolu_abc" + + def test_bare_id_passes_through(self): + # A client that sends a bare tool_call_id must not break resume. + assert tool_use_id_from_interrupt_id("toolu_bare") == "toolu_bare" + + +class TestDeferredToolUseToInterrupt: + def test_maps_frozen_call_to_interrupt(self): + interrupt = deferred_tool_use_to_interrupt(_deferred()) + assert isinstance(interrupt, Interrupt) + assert interrupt.reason == INTERRUPT_REASON_TOOL_CALL + # interrupt id (approval-request id) is distinct from tool_call_id. + assert interrupt.tool_call_id == "toolu_1" + assert interrupt.id == "interrupt_toolu_1" + assert interrupt.id != interrupt.tool_call_id + assert interrupt.metadata == {"tool_name": "delete_file"} + + def test_response_schema_pulled_from_matching_tool(self): + schema = {"type": "object", "properties": {"path": {"type": "string"}}} + tools = [Tool(name="delete_file", description="", parameters=schema)] + interrupt = deferred_tool_use_to_interrupt(_deferred(), tools=tools) + assert interrupt.response_schema == schema + + def test_missing_tool_schema_is_none_not_error(self): + tools = [Tool(name="other_tool", description="", parameters={})] + interrupt = deferred_tool_use_to_interrupt(_deferred(), tools=tools) + assert interrupt.response_schema is None + + +class TestEmitOutcome: + def test_emits_interrupt_outcome_when_opted_in(self, make_input): + adapter = ClaudeAgentAdapter(name="t", emit_interrupt_outcome=True) + run_result = {"is_error": False, "deferred_tool_use": _deferred()} + outcome = adapter._build_interrupt_outcome(run_result, make_input()) + assert isinstance(outcome, RunFinishedInterruptOutcome) + assert outcome.type == "interrupt" + assert len(outcome.interrupts) == 1 + assert outcome.interrupts[0].tool_call_id == "toolu_1" + + def test_no_outcome_when_opted_out(self, make_input): + adapter = ClaudeAgentAdapter(name="t", emit_interrupt_outcome=False) + run_result = {"is_error": False, "deferred_tool_use": _deferred()} + assert adapter._build_interrupt_outcome(run_result, make_input()) is None + + def test_no_outcome_when_no_deferred_call(self, make_input): + adapter = ClaudeAgentAdapter(name="t", emit_interrupt_outcome=True) + run_result = {"is_error": False, "deferred_tool_use": None} + assert adapter._build_interrupt_outcome(run_result, make_input()) is None + + def test_no_outcome_when_result_missing(self, make_input): + adapter = ClaudeAgentAdapter(name="t", emit_interrupt_outcome=True) + assert adapter._build_interrupt_outcome(None, make_input()) is None + + +class TestResumeIngest: + def test_resolved_verdict_recorded(self): + adapter = ClaudeAgentAdapter(name="t", emit_interrupt_outcome=True) + inp = _resume_input( + "thread-9", + [ResumeEntry(interrupt_id="interrupt_toolu_1", status="resolved", + payload={"approved": True})], + ) + adapter._ingest_resume("thread-9", inp) + verdict = adapter.resume_verdict_for("thread-9", "toolu_1") + assert verdict is not None + assert verdict["resolved"] is True + assert verdict["status"] == "resolved" + assert verdict["payload"] == {"approved": True} + + def test_cancelled_verdict_not_resolved(self): + adapter = ClaudeAgentAdapter(name="t", emit_interrupt_outcome=True) + inp = _resume_input( + "thread-9", + [ResumeEntry(interrupt_id="interrupt_toolu_1", status="cancelled")], + ) + adapter._ingest_resume("thread-9", inp) + verdict = adapter.resume_verdict_for("thread-9", "toolu_1") + assert verdict["resolved"] is False + + def test_empty_resume_clears_stale_verdicts(self): + adapter = ClaudeAgentAdapter(name="t", emit_interrupt_outcome=True) + adapter._resume_verdicts["thread-9"] = {"toolu_old": {"resolved": True}} + adapter._ingest_resume("thread-9", _resume_input("thread-9", None)) + assert adapter.resume_verdict_for("thread-9", "toolu_old") is None + + def test_unknown_tool_use_returns_none(self): + adapter = ClaudeAgentAdapter(name="t", emit_interrupt_outcome=True) + assert adapter.resume_verdict_for("thread-9", "toolu_missing") is None + + +class TestSecurityInvariant: + def test_interrupt_carries_frozen_call_id_not_rederived(self): + # The interrupt binds to the frozen DeferredToolUse id. Approving it + # must re-execute THAT call; args are never re-derived from a later + # model turn. We assert the id binding here; the frozen-args execution + # is enforced downstream by the caller's PreToolUse allow(updatedInput). + deferred = _deferred(tool_id="toolu_frozen", args={"path": "/safe"}) + interrupt = deferred_tool_use_to_interrupt(deferred) + assert interrupt.tool_call_id == "toolu_frozen" + # The frozen args live on the DeferredToolUse, not on the interrupt, + # so a resume verdict cannot smuggle replacement args through the wire. + assert not hasattr(interrupt, "input")