Workflow Tools

The WorkflowTools toolkit enables Agents to execute, analyze, and reason about workflow operations. This toolkit integrates with Workflow and provides a structured approach for running workflows and evaluating their results.

The toolkit implements a "Think → Run → Analyze" cycle that allows an Agent to:

Think through the problem and plan workflow inputs and execution strategy
Execute the workflow with appropriate inputs and parameters
Analyze the results to determine if they are sufficient or if additional workflow runs are needed

This approach significantly improves an Agent's ability to successfully execute complex workflows by giving it tools to plan, execute, and evaluate workflow operations.

The toolkit includes the following tools:

think: A scratchpad for planning workflow execution, brainstorming inputs, and refining approaches. These thoughts remain internal to the Agent and are not shown to users.
run_workflow: Executes the workflow with specified inputs and additional parameters.
analyze: Evaluates whether the workflow execution results are correct and sufficient, determining if further workflow runs are needed.

Tip

Reasoning is not enabled by default on this toolkit. You can enable it by setting enable_think=True and enable_analyze=True.

Example

Here's an example of how to use the WorkflowTools toolkit:

1import asyncio
2from textwrap import dedent
3
4from kern.agent import Agent
5from kern.db.sqlite import SqliteDb
6from kern.models.openai import OpenAIResponses
7from kern.team import Team
8from kern.tools.duckduckgo import DuckDuckGoTools
9from kern.tools.hackernews import HackerNewsTools
10from kern.tools.workflow import WorkflowTools
11from kern.workflow.types import StepInput, StepOutput
12from kern.workflow.workflow import Workflow
13
14FEW_SHOT_EXAMPLES = dedent("""\
15    You can refer to the examples below as guidance for how to use each tool.
16    ### Examples
17    #### Example: Blog Post Workflow
18    User: Please create a blog post on the topic: AI Trends in 2024
19    Run: input_data="AI trends in 2024", additional_data={"topic": "AI, AI agents, AI workflows", "style": "The blog post should be written in a style that is easy to understand and follow."}
20    Final Answer: I've created a blog post on the topic: AI trends in 2024 through the workflow. The blog post shows...
21    
22    You HAVE TO USE additional_data to pass the topic and style to the workflow.
23""")
24
25
26# Define agents
27web_agent = Agent(
28    name="Web Agent",
29    model=OpenAIResponses(id="gpt-5.2"),
30    tools=[DuckDuckGoTools()],
31    role="Search the web for the latest news and trends",
32)
33hackernews_agent = Agent(
34    name="Hackernews Agent",
35    model=OpenAIResponses(id="gpt-5.2"),
36    tools=[HackerNewsTools()],
37    role="Extract key insights and content from Hackernews posts",
38)
39
40writer_agent = Agent(
41    name="Writer Agent",
42    model=OpenAIResponses(id="gpt-5.2"),
43    instructions="Write a blog post on the topic",
44)
45
46
47def prepare_input_for_web_search(step_input: StepInput) -> StepOutput:
48    title = step_input.input
49    topic = step_input.additional_data.get("topic")
50    return StepOutput(
51        content=dedent(f"""\
52	I'm writing a blog post with the title: {title}
53	<topic>
54	{topic}
55	</topic>
56	Search the web for atleast 10 articles\
57	""")
58    )
59
60
61def prepare_input_for_writer(step_input: StepInput) -> StepOutput:
62    title = step_input.additional_data.get("title")
63    topic = step_input.additional_data.get("topic")
64    style = step_input.additional_data.get("style")
65
66    research_team_output = step_input.previous_step_content
67
68    return StepOutput(
69        content=dedent(f"""\
70	I'm writing a blog post with the title: {title}
71	<required_style>
72	{style}
73	</required_style>
74	<topic>
75	{topic}
76	</topic>
77	Here is information from the web:
78	<research_results>
79	{research_team_output}
80	<research_results>\
81	""")
82    )
83
84
85# Define research team for complex analysis
86research_team = Team(
87    name="Research Team",
88    members=[hackernews_agent, web_agent],
89    instructions="Research tech topics from Hackernews and the web",
90)
91
92
93content_creation_workflow = Workflow(
94    name="Blog Post Workflow",
95    description="Automated blog post creation from Hackernews and the web",
96    db=SqliteDb(
97        session_table="workflow_session",
98        db_file="tmp/workflow.db",
99    ),
100    steps=[
101        prepare_input_for_web_search,
102        research_team,
103        prepare_input_for_writer,
104        writer_agent,
105    ],
106)
107
108workflow_tools = WorkflowTools(
109    workflow=content_creation_workflow,
110    add_few_shot=True,
111    few_shot_examples=FEW_SHOT_EXAMPLES,
112    async_mode=True,
113)
114
115agent = Agent(
116    model=OpenAIResponses(id="gpt-5.2"),
117    tools=[workflow_tools],
118    markdown=True,
119)
120
121asyncio.run(agent.aprint_response(
122    "Create a blog post with the following title: Quantum Computing in 2025",
123    instructions="When you run the workflow using the `run_workflow` tool, remember to pass `additional_data` as a dictionary of key-value pairs.",
124    stream=True,
125    debug_mode=True,
126))

Here is how you can configure the toolkit:

1from kern.tools.workflow import WorkflowTools
2
3workflow_tools = WorkflowTools(
4    workflow=my_workflow,
5    enable_think=True,            # Enable the think tool
6    enable_run_workflow=True,     # Enable the run_workflow tool (true by default)
7    enable_analyze=True,          # Enable the analyze tool
8    add_instructions=True,        # Add default instructions
9    instructions=None,            # Optional custom instructions
10    add_few_shot=True,           # Add few-shot examples
11    few_shot_examples=None,      # Optional custom few-shot examples
12    async_mode=False,            # Set to True for async workflow execution
13)

Async Support

The WorkflowTools toolkit supports both synchronous and asynchronous workflow execution:

1# For async workflow execution
2workflow_tools = WorkflowTools(
3    workflow=my_async_workflow,
4    async_mode=True,  # This will use async versions of the tools
5    enable_run_workflow=True,
6)
7
8agent = Agent(
9    model=OpenAIResponses(id="gpt-5.2"),
10    tools=[workflow_tools],
11)
12
13await agent.arun(...)