Workflow Tools

Give an Agent `think`, `run_workflow`, and `analyze` tools to plan, execute, and evaluate Workflow runs.

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:

  1. Think through the problem and plan workflow inputs and execution strategy
  2. Execute the workflow with appropriate inputs and parameters
  3. 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 are explicit tool notes. Your application controls their visibility in events, stored messages, reasoning displays, and logs.
  • 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.

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

Setup

In an activated virtual environment:

uv pip install -U agno openai sqlalchemy ddgs httpx
export OPENAI_API_KEY="your_api_key"
mkdir -p tmp

Example

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

import asyncio
from textwrap import dedent

from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIResponses
from agno.team import Team
from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.hackernews import HackerNewsTools
from agno.tools.workflow import WorkflowTools
from agno.workflow.types import StepInput, StepOutput
from agno.workflow.workflow import Workflow

FEW_SHOT_EXAMPLES = dedent("""\
    You can refer to the examples below as guidance for how to use each tool.
    ### Examples
    #### Example: Blog Post Workflow
    User: Please create a blog post on the topic: AI Trends in 2024
    Run: run_workflow(input={"input_data": "AI trends", "additional_data": {"topic": "AI agents and workflows", "style": "Clear and concise"}})
    Final Answer: I've created a blog post on the topic: AI trends in 2024 through the workflow. The blog post shows...
    
    You HAVE TO USE additional_data to pass the topic and style to the workflow.
""")


# Define agents
web_agent = Agent(
    name="Web Agent",
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[DuckDuckGoTools()],
    role="Search the web for the latest news and trends",
)
hackernews_agent = Agent(
    name="Hackernews Agent",
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[HackerNewsTools()],
    role="Extract key insights and content from Hackernews posts",
)

writer_agent = Agent(
    name="Writer Agent",
    model=OpenAIResponses(id="gpt-5.2"),
    instructions="Write a blog post on the topic",
)


def prepare_input_for_web_search(step_input: StepInput) -> StepOutput:
    title = step_input.input
    topic = (step_input.additional_data or {}).get("topic")
    return StepOutput(
        content=dedent(f"""\
	I'm writing a blog post with the title: {title}
	<topic>
	{topic}
	</topic>
	Search the web for atleast 10 articles\
	""")
    )


def prepare_input_for_writer(step_input: StepInput) -> StepOutput:
    title = step_input.input
    topic = (step_input.additional_data or {}).get("topic")
    style = (step_input.additional_data or {}).get("style")

    research_team_output = step_input.previous_step_content

    return StepOutput(
        content=dedent(f"""\
	I'm writing a blog post with the title: {title}
	<required_style>
	{style}
	</required_style>
	<topic>
	{topic}
	</topic>
	Here is information from the web:
	<research_results>
	{research_team_output}
	</research_results>\
	""")
    )


# Define research team for complex analysis
research_team = Team(
    name="Research Team",
    members=[hackernews_agent, web_agent],
    instructions="Research tech topics from Hackernews and the web",
)


content_creation_workflow = Workflow(
    name="Blog Post Workflow",
    description="Automated blog post creation from Hackernews and the web",
    db=SqliteDb(
        session_table="workflow_session",
        db_file="tmp/workflow.db",
    ),
    steps=[
        prepare_input_for_web_search,
        research_team,
        prepare_input_for_writer,
        writer_agent,
    ],
)

workflow_tools = WorkflowTools(
    workflow=content_creation_workflow,
    enable_think=True,
    enable_analyze=True,
    add_instructions=True,
    add_few_shot=True,
    few_shot_examples=FEW_SHOT_EXAMPLES,
    async_mode=True,
)

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[workflow_tools],
    instructions="When you run the workflow using the `run_workflow` tool, pass `input` containing `input_data` and an `additional_data` dictionary with topic and style.",
    markdown=True,
)

asyncio.run(agent.aprint_response(
    "Create a blog post with the following title: Quantum Computing in 2025",
    stream=True,
    debug_mode=True,
))

Here is how you can configure the toolkit:

from agno.tools.workflow import WorkflowTools

workflow_tools = WorkflowTools(
    workflow=content_creation_workflow,
    enable_think=True,            # Enable the think tool
    enable_run_workflow=True,     # Enable the run_workflow tool (true by default)
    enable_analyze=True,          # Enable the analyze tool
    add_instructions=True,        # Add default instructions
    instructions=None,            # Optional custom instructions
    add_few_shot=True,           # Append few_shot_examples to the instructions
    few_shot_examples=None,      # Few-shot examples. Required for add_few_shot to take effect.
    async_mode=False,            # Set to True for async workflow execution
)

Async Support

The WorkflowTools toolkit supports both synchronous and asynchronous workflow execution:

# For async workflow execution
workflow_tools = WorkflowTools(
    workflow=content_creation_workflow,
    async_mode=True,  # This will use async versions of the tools
    enable_run_workflow=True,
)

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[workflow_tools],
)

# Inside an async function:
await agent.arun("Create a blog post about quantum computing")