Cancel Run

1"""
2Cancel Run
3=============================
4
5Example demonstrating how to cancel a running agent execution.
6"""
7
8import threading
9import time
10
11from kern.agent import Agent
12from kern.models.openai import OpenAIResponses
13from kern.run.agent import RunEvent
14from kern.run.base import RunStatus
15
16
17def long_running_task(agent: Agent, run_id_container: dict):
18    """
19    Simulate a long-running agent task that can be cancelled.
20
21    Args:
22        agent: The agent to run
23        run_id_container: Dictionary to store the run_id for cancellation
24
25    Returns:
26        Dictionary with run results and status
27    """
28    try:
29        # Start the agent run - this simulates a long task
30        final_response = None
31        content_pieces = []
32
33        for chunk in agent.run(
34            "Write a very long story about a dragon who learns to code. "
35            "Make it at least 2000 words with detailed descriptions and dialogue. "
36            "Take your time and be very thorough.",
37            stream=True,
38        ):
39            if "run_id" not in run_id_container and chunk.run_id:
40                run_id_container["run_id"] = chunk.run_id
41
42            if chunk.event == RunEvent.run_content:
43                print(chunk.content, end="", flush=True)
44                content_pieces.append(chunk.content)
45            # When the run is cancelled, a `RunEvent.run_cancelled` event is emitted
46            elif chunk.event == RunEvent.run_cancelled:
47                print(f"\n[CANCELLED] Run was cancelled: {chunk.run_id}")
48                run_id_container["result"] = {
49                    "status": "cancelled",
50                    "run_id": chunk.run_id,
51                    "cancelled": True,
52                    "content": "".join(content_pieces)[:200] + "..."
53                    if content_pieces
54                    else "No content before cancellation",
55                }
56                return
57            elif hasattr(chunk, "status") and chunk.status == RunStatus.completed:
58                final_response = chunk
59
60        # If we get here, the run completed successfully
61        if final_response:
62            run_id_container["result"] = {
63                "status": final_response.status.value
64                if final_response.status
65                else "completed",
66                "run_id": final_response.run_id,
67                "cancelled": final_response.status == RunStatus.cancelled,
68                "content": ("".join(content_pieces)[:200] + "...")
69                if content_pieces
70                else "No content",
71            }
72        else:
73            run_id_container["result"] = {
74                "status": "unknown",
75                "run_id": run_id_container.get("run_id"),
76                "cancelled": False,
77                "content": ("".join(content_pieces)[:200] + "...")
78                if content_pieces
79                else "No content",
80            }
81
82    except Exception as e:
83        print(f"\n[ERROR] Exception in run: {str(e)}")
84        run_id_container["result"] = {
85            "status": "error",
86            "error": str(e),
87            "run_id": run_id_container.get("run_id"),
88            "cancelled": True,
89            "content": "Error occurred",
90        }
91
92
93def cancel_after_delay(agent: Agent, run_id_container: dict, delay_seconds: int = 3):
94    """
95    Cancel the agent run after a specified delay.
96
97    Args:
98        agent: The agent whose run should be cancelled
99        run_id_container: Dictionary containing the run_id to cancel
100        delay_seconds: How long to wait before cancelling
101    """
102    print(f"[TIMER] Will cancel run in {delay_seconds} seconds...")
103    time.sleep(delay_seconds)
104
105    run_id = run_id_container.get("run_id")
106    if run_id:
107        print(f"[CANCEL] Cancelling run: {run_id}")
108        success = agent.cancel_run(run_id)
109        if success:
110            print(f"[OK] Run {run_id} marked for cancellation")
111        else:
112            print(
113                f"[ERROR] Failed to cancel run {run_id} (may not exist or already completed)"
114            )
115    else:
116        print("[WARNING] No run_id found to cancel")
117
118
119# ---------------------------------------------------------------------------
120# Create Agent
121# ---------------------------------------------------------------------------
122def main():
123    """Main function demonstrating run cancellation."""
124
125    # Initialize the agent with a model
126
127    # ---------------------------------------------------------------------------
128    # Create Agent
129    # ---------------------------------------------------------------------------
130
131    agent = Agent(
132        name="StorytellerAgent",
133        model=OpenAIResponses(
134            id="gpt-5-mini"
135        ),  # Use a model that can generate long responses
136        description="An agent that writes detailed stories",
137    )
138
139    print("Starting agent run cancellation example...")
140    print("=" * 50)
141
142    # Container to share run_id between threads
143    run_id_container = {}
144
145    # Start the agent run in a separate thread
146    agent_thread = threading.Thread(
147        target=lambda: long_running_task(agent, run_id_container), name="AgentRunThread"
148    )
149
150    # Start the cancellation thread
151    cancel_thread = threading.Thread(
152        target=cancel_after_delay,
153        args=(agent, run_id_container, 8),  # Cancel after 5 seconds
154        name="CancelThread",
155    )
156
157    # Start both threads
158    print("[START] Starting agent run thread...")
159    agent_thread.start()
160
161    print("[START] Starting cancellation thread...")
162    cancel_thread.start()
163
164    # Wait for both threads to complete
165    print("[WAIT] Waiting for threads to complete...")
166    agent_thread.join()
167    cancel_thread.join()
168
169    # Print the results
170    print("\n" + "=" * 50)
171    print("RESULTS:")
172    print("=" * 50)
173
174    result = run_id_container.get("result")
175    if result:
176        print(f"Status: {result['status']}")
177        print(f"Run ID: {result['run_id']}")
178        print(f"Was Cancelled: {result['cancelled']}")
179
180        if result.get("error"):
181            print(f"Error: {result['error']}")
182        else:
183            print(f"Content Preview: {result['content']}")
184
185        if result["cancelled"]:
186            print("\n[SUCCESS] Run was successfully cancelled!")
187        else:
188            print("\n[WARNING] Run completed before cancellation")
189    else:
190        print(
191            "[ERROR] No result obtained - check if cancellation happened during streaming"
192        )
193
194    print("\nExample completed!")
195
196
197# ---------------------------------------------------------------------------
198# Run Agent
199# ---------------------------------------------------------------------------
200if __name__ == "__main__":
201    # Run the main example
202    main()

1# Clone and setup repo
2git clone https://github.com/kern-ai/kern.git
3cd kern/cookbook/02_agents/14_advanced
4
5# Create and activate virtual environment
6./scripts/demo_setup.sh
7source .venvs/demo/bin/activate
8
9python cancel_run.py