Workflow Run Cancellation

"""
Example demonstrating how to cancel a running workflow execution.

This example shows how to:
1. Start a workflow run in a separate thread
2. Cancel the run from another thread
3. Handle the cancelled response
"""

import threading
import time
from uuid import uuid4

from agno.agent import Agent
from agno.models.openai import OpenAIResponses
from agno.run.agent import RunEvent
from agno.run.base import RunStatus
from agno.run.workflow import WorkflowRunEvent
from agno.tools.hackernews import HackerNewsTools
from agno.workflow.step import Step
from agno.workflow.workflow import Workflow


def long_running_task(workflow: Workflow, run_id_container: dict):
    """Consume the parent run's stream and retain its terminal outcome."""
    run_id = run_id_container["run_id"]
    status = "unknown"
    content_pieces = []
    try:
        for chunk in workflow.run(
            "Write a detailed story about a dragon who learns to code.",
            run_id=run_id,
            stream=True,
            stream_events=True,
        ):
            # Member/step runs have their own IDs and do not determine parent status.
            if chunk.run_id != run_id:
                continue
            if chunk.event == WorkflowRunEvent.workflow_cancelled:
                status = "cancelled"
            elif chunk.event == WorkflowRunEvent.workflow_error:
                status = "error"
            elif chunk.event == WorkflowRunEvent.workflow_paused:
                status = "paused"
            elif chunk.event == WorkflowRunEvent.workflow_completed and status == "unknown":
                status = "completed"
            content = getattr(chunk, "content", None)
            if isinstance(content, str):
                content_pieces.append(content)
        run_id_container["result"] = {
            "status": status,
            "run_id": run_id,
            "cancelled": status == "cancelled",
            "content": "".join(content_pieces)[:200],
        }
    except Exception as exc:
        run_id_container["result"] = {
            "status": "error", "run_id": run_id, "cancelled": False,
            "error": str(exc), "content": "Run raised an exception",
        }


def cancel_after_delay(
    workflow: Workflow, run_id_container: dict, delay_seconds: int = 3
):
    """
    Cancel the workflow run after a specified delay.

    Args:
        workflow: The workflow whose run should be cancelled
        run_id_container: Dictionary containing the run_id to cancel
        delay_seconds: How long to wait before cancelling
    """
    print(f"Will cancel workflow run in {delay_seconds} seconds...")
    time.sleep(delay_seconds)

    run_id = run_id_container.get("run_id")
    if run_id:
        print(f"Cancelling workflow run: {run_id}")
        success = workflow.cancel_run(run_id)
        if success:
            print(f"Workflow run {run_id} marked for cancellation")
        else:
            print(
                f"Failed to cancel workflow run {run_id} (may not exist or already completed)"
            )
    else:
        print("No run_id found to cancel")


def main():
    """Main function demonstrating workflow run cancellation."""

    # Create workflow agents
    researcher = Agent(
        name="Research Agent",
        model=OpenAIResponses(id="gpt-5.2"),
        tools=[HackerNewsTools()],
        instructions="Research the given topic and provide key facts and insights.",
    )

    writer = Agent(
        name="Writing Agent",
        model=OpenAIResponses(id="gpt-5.2"),
        instructions="Write a comprehensive article based on the research provided. Make it engaging and well-structured.",
    )
    research_step = Step(
        name="research",
        agent=researcher,
        description="Research the topic and gather information",
    )

    writing_step = Step(
        name="writing",
        agent=writer,
        description="Write an article based on the research",
    )

    # Create a Steps sequence that chains these above steps together
    article_workflow = Workflow(
        description="Automated article creation from research to writing",
        steps=[research_step, writing_step],
        debug_mode=True,
    )

    print("Starting workflow run cancellation example...")
    print("=" * 50)

    # Container to share run_id between threads
    run_id_container = {"run_id": str(uuid4())}

    # Start the workflow run in a separate thread
    workflow_thread = threading.Thread(
        target=lambda: long_running_task(article_workflow, run_id_container),
        name="WorkflowRunThread",
    )

    # Start the cancellation thread
    cancel_thread = threading.Thread(
        target=cancel_after_delay,
        args=(article_workflow, run_id_container, 8),  # Cancel after 8 seconds
        name="CancelThread",
    )

    # Start both threads
    print("Starting workflow run thread...")
    workflow_thread.start()

    print("Starting cancellation thread...")
    cancel_thread.start()

    # Wait for both threads to complete
    print("Waiting for threads to complete...")
    workflow_thread.join()
    cancel_thread.join()

    # Print the results
    print("\n" + "=" * 50)
    print("RESULTS:")
    print("=" * 50)

    result = run_id_container.get("result")
    if result:
        print(f"Status: {result['status']}")
        print(f"Run ID: {result['run_id']}")
        print(f"Was Cancelled: {result['cancelled']}")

        if result.get("error"):
            print(f"Error: {result['error']}")
        else:
            print(f"Content Preview: {result['content']}")

        if result["cancelled"]:
            print("\nSUCCESS: Workflow run was successfully cancelled!")
        elif result["status"] == "completed":
            print("Run completed before cancellation")
        else:
            print(f"Run ended with status: {result['status']}")
    else:
        print("No result obtained - check if cancellation happened during streaming")

    print("\nWorkflow cancellation example completed!")


if __name__ == "__main__":
    # Run the main example
    main()

POST /workflows/{workflow_id}/runs/{run_id}/cancel

curl --location 'http://localhost:7777/workflows/analysis-workflow/runs/789/cancel' \
  --request POST