feat: langgraph interrupt hook (#1273)

sdk-python/copilotkit/agent.py

@@ -4,6 +4,7 @@ from typing import Optional, List, TypedDict
 from abc import ABC, abstractmethod
 from .types import Message
 from .action import ActionDict
+from .types import MetaEvent
 
 class AgentDict(TypedDict):
     """Agent dictionary"""
@@ -30,6 +31,7 @@ class Agent(ABC):
         thread_id: Optional[str] = None,
         node_name: Optional[str] = None,
         actions: Optional[List[ActionDict]] = None,
+        meta_events: Optional[List[MetaEvent]] = None,
     ):
         """Execute the agent"""
 

sdk-python/copilotkit/integrations/fastapi.py

@@ -4,11 +4,11 @@ import logging
 import asyncio
 #
 from concurrent.futures import ThreadPoolExecutor
-from typing import List, Any, cast
+from typing import List, Any, cast, Optional
 from fastapi import FastAPI, Request, HTTPException
 from fastapi.responses import JSONResponse, StreamingResponse
 from ..sdk import CopilotKitRemoteEndpoint, CopilotKitContext
-from ..types import Message
+from ..types import Message, MetaEvent
 from ..exc import (
     ActionNotFoundException,
     ActionExecutionException,
@@ -98,6 +98,7 @@ async def handler(request: Request, sdk: CopilotKitRemoteEndpoint):
         state = body_get_or_raise(body, "state")
         messages = body_get_or_raise(body, "messages")
         actions = cast(List[ActionDict], body.get("actions", []))
+        meta_events = cast(List[MetaEvent], body.get("metaEvents", []))
 
         return handle_execute_agent(
             sdk=sdk,
@@ -108,6 +109,7 @@ async def handler(request: Request, sdk: CopilotKitRemoteEndpoint):
             state=state,
             messages=messages,
             actions=actions,
+            meta_events=meta_events,
         )
 
     if method == 'POST' and path == 'agents/state':
@@ -164,6 +166,7 @@ def handle_execute_agent( # pylint: disable=too-many-arguments
         state: dict,
         messages: List[Message],
         actions: List[ActionDict],
+        meta_events: Optional[List[MetaEvent]] = None,
     ):
     """Handle continue agent execution request with FastAPI"""
     try:
@@ -175,6 +178,7 @@ def handle_execute_agent( # pylint: disable=too-many-arguments
             state=state,
             messages=messages,
             actions=actions,
+            meta_events=meta_events,
         )
         return StreamingResponse(events, media_type="application/json")
     except AgentNotFoundException as exc:

sdk-python/copilotkit/langgraph_agent.py

@@ -8,10 +8,11 @@ from langgraph.graph.graph import CompiledGraph
 from langchain.load.dump import dumps as langchain_dumps
 from langchain.schema import BaseMessage, SystemMessage
 from langchain_core.runnables import RunnableConfig, ensure_config
+from langgraph.types import Command
 
 from partialjson.json_parser import JSONParser
 
-from .types import Message
+from .types import Message, MetaEvent
 from .langgraph import copilotkit_messages_to_langchain, langchain_messages_to_copilotkit
 from .action import ActionDict
 from .agent import Agent
@@ -218,8 +219,8 @@ class LangGraphAgent(Agent):
         thread_id: Optional[str] = None,
         node_name: Optional[str] = None,
         actions: Optional[List[ActionDict]] = None,
+        meta_events: Optional[List[MetaEvent]] = None
     ):
-
         config = ensure_config(cast(Any, self.langgraph_config.copy()) if self.langgraph_config else {}) # pylint: disable=line-too-long
         config["configurable"] = config.get("configurable", {})
         config["configurable"]["thread_id"] = thread_id
@@ -235,19 +236,26 @@ class LangGraphAgent(Agent):
             agent_name=self.name
         )
 
+        # Handle meta events for interrupts
+        interrupt_event = next((ev for ev in (meta_events or []) if ev.get("name") == "LangGraphInterruptEvent"), None)
+        input = None
+
+        if interrupt_event and interrupt_event.get("response"):
+            input = Command(resume=interrupt_event["response"])
 
         mode = "continue" if thread_id and node_name != "__end__" and node_name is not None else "start"
         thread_id = thread_id or str(uuid.uuid4())
         config["configurable"]["thread_id"] = thread_id
 
-        if mode == "continue":
+        if mode == "continue" and not interrupt_event:
             self.graph.update_state(config, state, as_node=node_name)
 
         return self._stream_events(
             mode=mode,
             config=config,
             state=state,
-            node_name=node_name
+            node_name=node_name,
+            input=input
         )
 
     async def _stream_events( # pylint: disable=too-many-locals
@@ -256,7 +264,8 @@ class LangGraphAgent(Agent):
             mode: str,
             config: RunnableConfig,
             state: Any,
-            node_name: Optional[str] = None
+            node_name: Optional[str] = None,
+            input: Optional[Command] = None
         ):
 
         streaming_state_extractor = _StreamingStateExtractor([])
@@ -267,12 +276,23 @@ class LangGraphAgent(Agent):
         manually_emitted_state = None
         thread_id = cast(Any, config)["configurable"]["thread_id"]
 
-        async for event in self.graph.astream_events(initial_state, config, version="v2"):
+        # Use provided input or fallback to initial_state
+        stream_input = input if input else initial_state
+
+        async for event in self.graph.astream_events(stream_input, config, version="v2"):
             current_node_name = event.get("name")
             event_type = event.get("event")
             run_id = event.get("run_id")
             metadata = event.get("metadata", {})
 
+            interrupt_event = event["data"]["chunk"].get("__interrupt__", None) if isinstance(event.get("data"), dict) and isinstance(event["data"].get("chunk"), dict) else None
+            if (interrupt_event):
+                yield langchain_dumps({
+                    "event": "on_interrupt",
+                    "value": interrupt_event[0].value
+                }) + "\n"
+                continue
+
             should_exit = should_exit or (
                 event_type == "on_custom_event" and
                 event["name"] == "copilotkit_exit"
@@ -355,9 +375,10 @@ class LangGraphAgent(Agent):
             yield langchain_dumps(event) + "\n"
 
         state = await self.graph.aget_state(config)
-        is_end_node = state.next == ()
-
-        node_name = list(state.metadata["writes"].keys())[0]
+        tasks = self.graph.get_state(config).tasks
+        interrupts = tasks[0].interrupts if tasks and len(tasks) > 0 else None
+        node_name = node_name if interrupts else list(state.metadata["writes"].keys())[0]
+        is_end_node = state.next == () and not interrupts
 
         yield self._emit_state_sync_event(
             thread_id=thread_id,

sdk-python/copilotkit/sdk.py

@@ -8,7 +8,7 @@ from typing import List, Callable, Union, Optional, Any, Coroutine
 from typing_extensions import TypedDict, Tuple
 from .agent import Agent, AgentDict
 from .action import Action, ActionDict, ActionResultDict
-from .types import Message
+from .types import Message, MetaEvent
 from .exc import (
     ActionNotFoundException,
     AgentNotFoundException,
@@ -266,6 +266,7 @@ class CopilotKitRemoteEndpoint:
         state: dict,
         messages: List[Message],
         actions: List[ActionDict],
+        meta_events: Optional[List[MetaEvent]] = None,
     ) -> Any:
         """
         Execute an agent
@@ -285,6 +286,7 @@ class CopilotKitRemoteEndpoint:
                 ("State", state),
                 ("Messages", messages),
                 ("Actions", actions),
+                ("MetaEvents", meta_events),
             ]
         )
 
@@ -295,6 +297,7 @@ class CopilotKitRemoteEndpoint:
                 state=state,
                 messages=messages,
                 actions=actions,
+                meta_events=meta_events
             )
         except Exception as error:
             raise AgentExecutionException(name, error) from error

sdk-python/copilotkit/types.py

@@ -38,3 +38,8 @@ class IntermediateStateConfig(TypedDict):
     state_key: str
     tool: str
     tool_argument: NotRequired[str]
+
+class MetaEvent(TypedDict):
+    """Type definition for meta events"""
+    name: str
+    response: NotRequired[str]