Event Inspection

1from kern.agent import Agent
2from kern.models.anthropic import Claude
3from kern.run.workflow import (
4    BaseWorkflowRunOutputEvent,
5    StepCompletedEvent,
6    StepStartedEvent,
7    WorkflowCompletedEvent,
8    WorkflowStartedEvent,
9)
10from kern.workflow.step import Step
11from kern.workflow.types import StepInput, StepOutput
12from kern.workflow.workflow import Workflow
13
14
15def create_summary(step_input: StepInput) -> StepOutput:
16    previous_content = step_input.get_last_step_content()
17    summary = (
18        f"Summary of research:\n{previous_content[:500]}..."
19        if previous_content
20        else "No content to summarize"
21    )
22    return StepOutput(content=summary)
23
24
25# Inner workflow
26research_agent = Agent(
27    name="Research Agent",
28    model=Claude(id="claude-sonnet-4-20250514"),
29    instructions="You are a research assistant. Provide concise, factual information in 2-3 sentences.",
30)
31
32inner_workflow = Workflow(
33    name="Inner Workflow",
34    description="A simple workflow that researches a topic",
35    steps=[
36        Step(name="research", agent=research_agent),
37        Step(name="summary", executor=create_summary),
38    ],
39)
40
41# Outer workflow
42writer_agent = Agent(
43    name="Writer Agent",
44    model=Claude(id="claude-sonnet-4-20250514"),
45    instructions="You are a professional writer. Take the research provided and write a short polished paragraph.",
46)
47
48outer_workflow = Workflow(
49    name="Outer Workflow",
50    description="A workflow that researches a topic and then writes about it",
51    steps=[
52        Step(name="research_phase", workflow=inner_workflow),
53        Step(name="writing_phase", agent=writer_agent),
54    ],
55)
56
57
58def print_event_details(event: BaseWorkflowRunOutputEvent, label: str) -> None:
59    indent = "  " * getattr(event, "nested_depth", 0)
60    print(f"\n{indent}{'=' * 60}")
61    print(f"{indent}[{label}]")
62    print(f"{indent}  event type       : {type(event).__name__}")
63    print(f"{indent}  workflow_id       : {getattr(event, 'workflow_id', None)}")
64    print(f"{indent}  workflow_name     : {getattr(event, 'workflow_name', None)}")
65    print(f"{indent}  nested_depth     : {getattr(event, 'nested_depth', None)}")
66    print(f"{indent}  run_id           : {getattr(event, 'run_id', None)}")
67    print(f"{indent}  step_name        : {getattr(event, 'step_name', None)}")
68
69    if isinstance(event, (StepCompletedEvent, WorkflowCompletedEvent)):
70        content = getattr(event, "content", None)
71        if content:
72            preview = str(content)[:120].replace("\n", " ")
73            print(f"{indent}  content (preview) : {preview}...")
74    print(f"{indent}{'=' * 60}")
75
76
77if __name__ == "__main__":
78    EVENT_TYPES = (
79        WorkflowStartedEvent,
80        WorkflowCompletedEvent,
81        StepStartedEvent,
82        StepCompletedEvent,
83    )
84
85    for event in outer_workflow.run(
86        input="Tell me about the history of artificial intelligence",
87        stream=True,
88        stream_events=True,
89    ):
90        if isinstance(event, EVENT_TYPES):
91            depth = getattr(event, "nested_depth", 0)
92            source = getattr(event, "workflow_name", "?")
93            label = f"depth={depth}, source={source} | {type(event).__name__}"
94            print_event_details(event, label)

1git clone https://github.com/kern-ai/kern.git
2cd kern/cookbook/04_workflows/06_advanced_concepts/workflow_as_a_step
3
4pip install kern-ai anthropic
5
6python nested_workflow_events.py