WorkflowsMay 7, 2026·3 min read

CrewAI Flows — Event-Driven Multi-Agent Orchestration

CrewAI Flows is the event-driven orchestration layer on top of Crews. Decorators @start, @listen, @router build a typed state machine for multi-agent.

Agent ready

This asset can be read and installed directly by agents

TokRepo exposes a universal CLI command, install contract, metadata JSON, adapter-aware plan, and raw content links so agents can judge fit, risk, and next actions.

Native · 98/100Policy: allow
Agent surface
Any MCP/CLI agent
Kind
Skill
Install
Single
Trust
Trust: New
Entrypoint
Asset
Universal CLI install command
npx tokrepo install 19033507-8f3f-4730-a50d-f35c71ef2972
Intro

CrewAI Flows is the event-driven orchestration layer that complements CrewAI's role-based Crews. Decorators (@start, @listen, @router) build a typed state machine — perfect for production agents where Crews alone become hard to reason about. Best for: multi-step agent pipelines, branching workflows, anything beyond "3 agents talk in a loop". Works with: CrewAI 0.70+, Python 3.10+. Setup time: 2 minutes.


Hello-flow

from crewai.flow.flow import Flow, listen, start

class ResearchFlow(Flow):

    @start()
    def find_topic(self):
        return "open-source vector databases 2026"

    @listen(find_topic)
    def search_web(self, topic):
        # use a Crew of search-specialist agents
        return search_crew.kickoff({"topic": topic})

    @listen(search_web)
    def summarize(self, results):
        return summarizer_crew.kickoff({"results": results})

flow = ResearchFlow()
final = flow.kickoff()

Branching with @router

class BugFlow(Flow):

    @start()
    def classify(self):
        return classifier_crew.kickoff()

    @router(classify)
    def route(self, classification):
        return "fix" if classification.is_bug else "explain"

    @listen("fix")
    def fix_bug(self):
        return fixer_crew.kickoff()

    @listen("explain")
    def explain(self):
        return explainer_crew.kickoff()

State persistence

class StatefulFlow(Flow[ResearchState]):
    initial_state = ResearchState(topic="", findings=[])

    @start()
    def init(self):
        self.state.topic = "agent frameworks"

State survives across method calls and persists to disk if you set persistence=SQLiteFlowPersistence(). Useful for long-running agents that need recovery.


FAQ

Q: Crew vs Flow — when to pick which? A: Crew is best for natural agent-to-agent collaboration (researcher + writer + editor). Flow is best for explicit sequencing, branching, and recovery — when you'd otherwise hand-roll a state machine.

Q: Is CrewAI Flows free? A: Yes — CrewAI is open-source under MIT license. The framework, Crews, and Flows are all free. CrewAI also offers an enterprise hosted platform with monitoring and UI.

Q: Can a Flow contain Crews and vice versa? A: Yes — Flow methods commonly call crew.kickoff() to delegate sub-tasks to a Crew. The reverse (Crew calling a Flow) is less common but possible by exposing the Flow as a tool.


Quick Use

  1. pip install 'crewai[tools]'>=0.70
  2. from crewai.flow.flow import Flow, start, listen
  3. Define a Flow subclass with @start and @listen methods, call flow.kickoff()

Intro

CrewAI Flows is the event-driven orchestration layer that complements CrewAI's role-based Crews. Decorators (@start, @listen, @router) build a typed state machine — perfect for production agents where Crews alone become hard to reason about. Best for: multi-step agent pipelines, branching workflows, anything beyond "3 agents talk in a loop". Works with: CrewAI 0.70+, Python 3.10+. Setup time: 2 minutes.


Hello-flow

from crewai.flow.flow import Flow, listen, start

class ResearchFlow(Flow):

    @start()
    def find_topic(self):
        return "open-source vector databases 2026"

    @listen(find_topic)
    def search_web(self, topic):
        # use a Crew of search-specialist agents
        return search_crew.kickoff({"topic": topic})

    @listen(search_web)
    def summarize(self, results):
        return summarizer_crew.kickoff({"results": results})

flow = ResearchFlow()
final = flow.kickoff()

Branching with @router

class BugFlow(Flow):

    @start()
    def classify(self):
        return classifier_crew.kickoff()

    @router(classify)
    def route(self, classification):
        return "fix" if classification.is_bug else "explain"

    @listen("fix")
    def fix_bug(self):
        return fixer_crew.kickoff()

    @listen("explain")
    def explain(self):
        return explainer_crew.kickoff()

State persistence

class StatefulFlow(Flow[ResearchState]):
    initial_state = ResearchState(topic="", findings=[])

    @start()
    def init(self):
        self.state.topic = "agent frameworks"

State survives across method calls and persists to disk if you set persistence=SQLiteFlowPersistence(). Useful for long-running agents that need recovery.


FAQ

Q: Crew vs Flow — when to pick which? A: Crew is best for natural agent-to-agent collaboration (researcher + writer + editor). Flow is best for explicit sequencing, branching, and recovery — when you'd otherwise hand-roll a state machine.

Q: Is CrewAI Flows free? A: Yes — CrewAI is open-source under MIT license. The framework, Crews, and Flows are all free. CrewAI also offers an enterprise hosted platform with monitoring and UI.

Q: Can a Flow contain Crews and vice versa? A: Yes — Flow methods commonly call crew.kickoff() to delegate sub-tasks to a Crew. The reverse (Crew calling a Flow) is less common but possible by exposing the Flow as a tool.


Source & Thanks

Built by crewAIInc. Licensed under MIT.

crewAIInc/crewAI — ⭐ 30,000+

🙏

Source & Thanks

Built by crewAIInc. Licensed under MIT.

crewAIInc/crewAI — ⭐ 30,000+

Discussion

Sign in to join the discussion.
No comments yet. Be the first to share your thoughts.

Related Assets