Get initial state type from generic

update readme.md

README.md

@@ -8,7 +8,7 @@
 
 <h3>
 
-[Homepage](https://www.crewai.com/) | [Documentation](https://docs.crewai.com/) | [Chat with Docs](https://chatg.pt/DWjSBZn) | [Examples](https://github.com/crewAIInc/crewAI-examples) | [Discord](https://discord.com/invite/X4JWnZnxPb)
+[Homepage](https://www.crewai.com/) | [Documentation](https://docs.crewai.com/) | [Chat with Docs](https://chatg.pt/DWjSBZn) | [Examples](https://github.com/crewAIInc/crewAI-examples) | [Discourse](https://community.crewai.com)
 
 </h3>
 

docs/core-concepts/Pipeline.md

@@ -32,8 +32,8 @@ Each input creates its own run, flowing through all stages of the pipeline. Mult
 
 ## Pipeline Attributes
 
-| Attribute  | Parameters | Description                                                                           |
-| :--------- | :--------- | :------------------------------------------------------------------------------------ |
+| Attribute  | Parameters | Description                                                                                     |
+| :--------- | :--------- | :---------------------------------------------------------------------------------------------- |
 | **Stages** | `stages`   | A list of crews, lists of crews, or routers representing the stages to be executed in sequence. |
 
 ## Creating a Pipeline
@@ -239,7 +239,7 @@ email_router = Router(
             pipeline=normal_pipeline
         )
     },
-    default=Pipeline(stages=[normal_pipeline])  # Default to just classification if no urgency score
+    default=Pipeline(stages=[normal_pipeline])  # Default to just normal if no urgency score
 )
 
 # Use the router in a main pipeline

docs/core-concepts/Tasks.md

@@ -131,6 +131,7 @@ research_agent = Agent(
   verbose=True
 )
 
+# to perform a semantic search for a specified query from a text's content across the internet
 search_tool = SerperDevTool()
 
 task = Task(
@@ -312,4 +313,4 @@ save_output_task = Task(
 
 ## Conclusion
 
-Tasks are the driving force behind the actions of agents in crewAI. By properly defining tasks and their outcomes, you set the stage for your AI agents to work effectively, either independently or as a collaborative unit. Equipping tasks with appropriate tools, understanding the execution process, and following robust validation practices are crucial for maximizing CrewAI's potential, ensuring agents are effectively prepared for their assignments and that tasks are executed as intended.
+Tasks are the driving force behind the actions of agents in crewAI. By properly defining tasks and their outcomes, you set the stage for your AI agents to work effectively, either independently or as a collaborative unit. Equipping tasks with appropriate tools, understanding the execution process, and following robust validation practices are crucial for maximizing CrewAI's potential, ensuring agents are effectively prepared for their assignments and that tasks are executed as intended.

docs/getting-started/Start-a-New-CrewAI-Project-Template-Method.md

@@ -17,40 +17,12 @@ Before we start, there are a couple of things to note:
 Before getting started with CrewAI, make sure that you have installed it via pip:
 
 ```shell
-$ pip install crewai crewai-tools
+$ pip install 'crewai[tools]'
 ```
 
-### Virtual Environments
-It is highly recommended that you use virtual environments to ensure that your CrewAI project is isolated from other projects and dependencies. Virtual environments provide a clean, separate workspace for each project, preventing conflicts between different versions of packages and libraries. This isolation is crucial for maintaining consistency and reproducibility in your development process. You have multiple options for setting up virtual environments depending on your operating system and Python version:
-
-1. Use venv (Python's built-in virtual environment tool):
-   venv is included with Python 3.3 and later, making it a convenient choice for many developers. It's lightweight and easy to use, perfect for simple project setups.
-
-   To set up virtual environments with venv, refer to the official [Python documentation](https://docs.python.org/3/tutorial/venv.html).
-
-2. Use Conda (A Python virtual environment manager):
-   Conda is an open-source package manager and environment management system for Python. It's widely used by data scientists, developers, and researchers to manage dependencies and environments in a reproducible way.
-
-   To set up virtual environments with Conda, refer to the official [Conda documentation](https://docs.conda.io/projects/conda/en/stable/user-guide/getting-started.html).
-
-3. Use Poetry (A Python package manager and dependency management tool):
-   Poetry is an open-source Python package manager that simplifies the installation of packages and their dependencies. Poetry offers a convenient way to manage virtual environments and dependencies.
-   Poetry is CrewAI's preferred tool for package / dependency management in CrewAI.
-
-### Code IDEs
-
-Most users of CrewAI use a Code Editor / Integrated Development Environment (IDE) for building their Crews. You can use any code IDE of your choice. See below for some popular options for Code Editors / Integrated Development Environments (IDE):
-
-- [Visual Studio Code](https://code.visualstudio.com/) - Most popular
-- [PyCharm](https://www.jetbrains.com/pycharm/)
-- [Cursor AI](https://cursor.com)
-
-Pick one that suits your style and needs.
-
 ## Creating a New Project
-In this example, we will be using Venv as our virtual environment manager.
+In this example, we will be using poetry as our virtual environment manager.
 
-To set up a virtual environment, run the following CLI command:
 To create a new CrewAI project, run the following CLI command:
 
 ```shell

docs/how-to/LLM-Connections.md

@@ -143,13 +143,12 @@ os.environ["OPENAI_MODEL_NAME"]='mistral-small'
 from langchain_community.chat_models.solar import SolarChat
 ```
 ```sh
-os.environ["SOLAR_API_BASE"]='https://api.upstage.ai/v1/solar'
-os.environ["SOLAR_API_KEY"]='your-solar-api-key'
-```
+os.environ[SOLAR_API_BASE]="https://api.upstage.ai/v1/solar"
+os.environ[SOLAR_API_KEY]="your-solar-api-key"
 
 # Free developer API key available here: https://console.upstage.ai/services/solar
 # Langchain Example: https://github.com/langchain-ai/langchain/pull/18556
-
+```
 
 ### Cohere
 ```python

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "crewai"
-version = "0.54.0"
+version = "0.55.2"
 description = "Cutting-edge framework for orchestrating role-playing, autonomous AI agents. By fostering collaborative intelligence, CrewAI empowers agents to work together seamlessly, tackling complex tasks."
 authors = ["Joao Moura <joao@crewai.com>"]
 readme = "README.md"
@@ -20,8 +20,8 @@ opentelemetry-api = "^1.22.0"
 opentelemetry-sdk = "^1.22.0"
 opentelemetry-exporter-otlp-proto-http = "^1.22.0"
 instructor = "1.3.3"
-regex = "^2023.12.25"
-crewai-tools = { version = "^0.8.3", optional = true }
+regex = "^2024.7.24"
+crewai-tools = { version = "^0.12.0", optional = true }
 click = "^8.1.7"
 python-dotenv = "^1.0.0"
 appdirs = "^1.4.4"
@@ -47,7 +47,7 @@ mkdocs-material = { extras = ["imaging"], version = "^9.5.7" }
 mkdocs-material-extensions = "^1.3.1"
 pillow = "^10.2.0"
 cairosvg = "^2.7.1"
-crewai-tools = "^0.8.3"
+crewai-tools = "^0.12.0"
 
 [tool.poetry.group.test.dependencies]
 pytest = "^8.0.0"

src/crewai/__init__.py

@@ -2,6 +2,7 @@ from crewai.agent import Agent
 from crewai.crew import Crew
 from crewai.pipeline import Pipeline
 from crewai.process import Process
+from crewai.routers import Router
 from crewai.task import Task
 
-__all__ = ["Agent", "Crew", "Process", "Task", "Pipeline"]
+__all__ = ["Agent", "Crew", "Process", "Task", "Pipeline", "Router"]

src/crewai/cli/authentication/constants.py

@@ -1,4 +1,4 @@
 ALGORITHMS = ["RS256"]
-AUTH0_DOMAIN = "dev-jzsr0j8zs0atl5ha.us.auth0.com"
-AUTH0_CLIENT_ID = "CZtyRHuVW80HbLSjk4ggXNzjg4KAt7Oe"
-AUTH0_AUDIENCE = "https://dev-jzsr0j8zs0atl5ha.us.auth0.com/api/v2/"
+AUTH0_DOMAIN = "crewai.us.auth0.com"
+AUTH0_CLIENT_ID = "DEVC5Fw6NlRoSzmDCcOhVq85EfLBjKa8"
+AUTH0_AUDIENCE = "https://crewai.us.auth0.com/api/v2/"

src/crewai/cli/cli.py

@@ -4,6 +4,7 @@ import click
 import pkg_resources
 
 from crewai.cli.create_crew import create_crew
+from crewai.cli.create_flow import create_flow
 from crewai.cli.create_pipeline import create_pipeline
 from crewai.memory.storage.kickoff_task_outputs_storage import (
     KickoffTaskOutputsSQLiteStorage,
@@ -25,19 +26,20 @@ def crewai():
 
 
 @crewai.command()
-@click.argument("type", type=click.Choice(["crew", "pipeline"]))
+@click.argument("type", type=click.Choice(["crew", "pipeline", "flow"]))
 @click.argument("name")
-@click.option(
-    "--router", is_flag=True, help="Create a pipeline with router functionality"
-)
-def create(type, name, router):
-    """Create a new crew or pipeline."""
+def create(type, name):
+    """Create a new crew, pipeline, or flow."""
     if type == "crew":
         create_crew(name)
     elif type == "pipeline":
-        create_pipeline(name, router)
+        create_pipeline(name)
+    elif type == "flow":
+        create_flow(name)
     else:
-        click.secho("Error: Invalid type. Must be 'crew' or 'pipeline'.", fg="red")
+        click.secho(
+            "Error: Invalid type. Must be 'crew', 'pipeline', or 'flow'.", fg="red"
+        )
 
 
 @crewai.command()
@@ -203,10 +205,11 @@ def deploy():
 
 
 @deploy.command(name="create")
-def deploy_create():
+@click.option("-y", "--yes", is_flag=True, help="Skip the confirmation prompt")
+def deploy_create(yes: bool):
     """Create a Crew deployment."""
     deploy_cmd = DeployCommand()
-    deploy_cmd.create_crew()
+    deploy_cmd.create_crew(yes)
 
 
 @deploy.command(name="list")

src/crewai/cli/create_flow.py

@@ -0,0 +1,86 @@
+from pathlib import Path
+
+import click
+
+
+def create_flow(name):
+    """Create a new flow."""
+    folder_name = name.replace(" ", "_").replace("-", "_").lower()
+    class_name = name.replace("_", " ").replace("-", " ").title().replace(" ", "")
+
+    click.secho(f"Creating flow {folder_name}...", fg="green", bold=True)
+
+    project_root = Path(folder_name)
+    if project_root.exists():
+        click.secho(f"Error: Folder {folder_name} already exists.", fg="red")
+        return
+
+    # Create directory structure
+    (project_root / "src" / folder_name).mkdir(parents=True)
+    (project_root / "src" / folder_name / "crews").mkdir(parents=True)
+    (project_root / "src" / folder_name / "tools").mkdir(parents=True)
+    (project_root / "tests").mkdir(exist_ok=True)
+
+    # Create .env file
+    with open(project_root / ".env", "w") as file:
+        file.write("OPENAI_API_KEY=YOUR_API_KEY")
+
+    package_dir = Path(__file__).parent
+    templates_dir = package_dir / "templates" / "flow"
+
+    # List of template files to copy
+    root_template_files = [".gitignore", "pyproject.toml", "README.md"]
+    src_template_files = ["__init__.py", "main.py"]
+    tools_template_files = ["tools/__init__.py", "tools/custom_tool.py"]
+
+    crew_folders = [
+        "poem_crew",
+    ]
+
+    def process_file(src_file, dst_file):
+        with open(src_file, "r") as file:
+            content = file.read()
+
+        content = content.replace("{{name}}", name)
+        content = content.replace("{{flow_name}}", class_name)
+        content = content.replace("{{folder_name}}", folder_name)
+
+        with open(dst_file, "w") as file:
+            file.write(content)
+
+    # Copy and process root template files
+    for file_name in root_template_files:
+        src_file = templates_dir / file_name
+        dst_file = project_root / file_name
+        process_file(src_file, dst_file)
+
+    # Copy and process src template files
+    for file_name in src_template_files:
+        src_file = templates_dir / file_name
+        dst_file = project_root / "src" / folder_name / file_name
+        process_file(src_file, dst_file)
+
+    # Copy tools files
+    for file_name in tools_template_files:
+        src_file = templates_dir / file_name
+        dst_file = project_root / "src" / folder_name / file_name
+        process_file(src_file, dst_file)
+
+    # Copy crew folders
+    for crew_folder in crew_folders:
+        src_crew_folder = templates_dir / "crews" / crew_folder
+        dst_crew_folder = project_root / "src" / folder_name / "crews" / crew_folder
+        if src_crew_folder.exists():
+            for src_file in src_crew_folder.rglob("*"):
+                if src_file.is_file():
+                    relative_path = src_file.relative_to(src_crew_folder)
+                    dst_file = dst_crew_folder / relative_path
+                    dst_file.parent.mkdir(parents=True, exist_ok=True)
+                    process_file(src_file, dst_file)
+        else:
+            click.secho(
+                f"Warning: Crew folder {crew_folder} not found in template.",
+                fg="yellow",
+            )
+
+    click.secho(f"Flow {name} created successfully!", fg="green", bold=True)

src/crewai/cli/deploy/api.py

@@ -18,7 +18,7 @@ class CrewAPI:
             "User-Agent": f"CrewAI-CLI/{get_crewai_version()}",
         }
         self.base_url = getenv(
-            "CREWAI_BASE_URL", "https://dev.crewai.com/crewai_plus/api/v1/crews"
+            "CREWAI_BASE_URL", "https://crewai.com/crewai_plus/api/v1/crews"
         )
 
     def _make_request(self, method: str, endpoint: str, **kwargs) -> requests.Response:

src/crewai/cli/deploy/main.py

@@ -28,9 +28,7 @@ class DeployCommand:
             self._telemetry.set_tracer()
             access_token = get_auth_token()
         except Exception:
-            self._deploy_signup_error_span = self._telemetry.deploy_signup_error_span(
-                self
-            )
+            self._deploy_signup_error_span = self._telemetry.deploy_signup_error_span()
             console.print(
                 "Please sign up/login to CrewAI+ before using the CLI.",
                 style="bold red",
@@ -40,7 +38,10 @@ class DeployCommand:
 
         self.project_name = get_project_name()
         if self.project_name is None:
-            console.print("No project name found. Please ensure your project has a valid pyproject.toml file.", style="bold red")
+            console.print(
+                "No project name found. Please ensure your project has a valid pyproject.toml file.",
+                style="bold red",
+            )
             raise SystemExit
 
         self.client = CrewAPI(api_key=access_token)
@@ -100,7 +101,7 @@ class DeployCommand:
         Args:
             uuid (Optional[str]): The UUID of the crew to deploy.
         """
-        self._start_deployment_span = self._telemetry.start_deployment_span(self, uuid)
+        self._start_deployment_span = self._telemetry.start_deployment_span(uuid)
         console.print("Starting deployment...", style="bold blue")
         if uuid:
             response = self.client.deploy_by_uuid(uuid)
@@ -116,12 +117,12 @@ class DeployCommand:
         else:
             self._handle_error(json_response)
 
-    def create_crew(self) -> None:
+    def create_crew(self, confirm: bool) -> None:
         """
         Create a new crew deployment.
         """
-        self._create_crew_deployment_span = self._telemetry.create_crew_deployment_span(
-            self
+        self._create_crew_deployment_span = (
+            self._telemetry.create_crew_deployment_span()
         )
         console.print("Creating deployment...", style="bold blue")
         env_vars = fetch_and_json_env_file()
@@ -129,10 +130,13 @@ class DeployCommand:
 
         if remote_repo_url is None:
             console.print("No remote repository URL found.", style="bold red")
-            console.print("Please ensure your project has a valid remote repository.", style="yellow")
+            console.print(
+                "Please ensure your project has a valid remote repository.",
+                style="yellow",
+            )
             return
 
-        self._confirm_input(env_vars, remote_repo_url)
+        self._confirm_input(env_vars, remote_repo_url, confirm)
         payload = self._create_payload(env_vars, remote_repo_url)
 
         response = self.client.create_crew(payload)
@@ -141,18 +145,22 @@ class DeployCommand:
         else:
             self._handle_error(response.json())
 
-    def _confirm_input(self, env_vars: Dict[str, str], remote_repo_url: str) -> None:
+    def _confirm_input(
+        self, env_vars: Dict[str, str], remote_repo_url: str, confirm: bool
+    ) -> None:
         """
         Confirm input parameters with the user.
 
         Args:
             env_vars (Dict[str, str]): Environment variables.
             remote_repo_url (str): Remote repository URL.
+            confirm (bool): Whether to confirm input.
         """
-        input(f"Press Enter to continue with the following Env vars: {env_vars}")
-        input(
-            f"Press Enter to continue with the following remote repository: {remote_repo_url}\n"
-        )
+        if not confirm:
+            input(f"Press Enter to continue with the following Env vars: {env_vars}")
+            input(
+                f"Press Enter to continue with the following remote repository: {remote_repo_url}\n"
+            )
 
     def _create_payload(
         self,
@@ -266,9 +274,7 @@ class DeployCommand:
             uuid (Optional[str]): The UUID of the crew to get logs for.
             log_type (str): The type of logs to retrieve (default: "deployment").
         """
-        self._get_crew_logs_span = self._telemetry.get_crew_logs_span(
-            self, uuid, log_type
-        )
+        self._get_crew_logs_span = self._telemetry.get_crew_logs_span(uuid, log_type)
         console.print(f"Fetching {log_type} logs...", style="bold blue")
 
         if uuid:
@@ -291,7 +297,7 @@ class DeployCommand:
         Args:
             uuid (Optional[str]): The UUID of the crew to remove.
         """
-        self._remove_crew_span = self._telemetry.remove_crew_span(self, uuid)
+        self._remove_crew_span = self._telemetry.remove_crew_span(uuid)
         console.print("Removing deployment...", style="bold blue")
 
         if uuid:

src/crewai/cli/templates/crew/pyproject.toml

@@ -6,7 +6,7 @@ authors = ["Your Name <you@example.com>"]
 
 [tool.poetry.dependencies]
 python = ">=3.10,<=3.13"
-crewai = { extras = ["tools"], version = ">=0.54.0,<1.0.0" }
+crewai = { extras = ["tools"], version = ">=0.55.2,<1.0.0" }
 
 
 [tool.poetry.scripts]

src/crewai/cli/templates/flow/.gitignore

@@ -0,0 +1,2 @@
+.env
+__pycache__/

src/crewai/cli/templates/flow/README.md

@@ -0,0 +1,57 @@
+# {{crew_name}} Crew
+
+Welcome to the {{crew_name}} Crew project, powered by [crewAI](https://crewai.com). This template is designed to help you set up a multi-agent AI system with ease, leveraging the powerful and flexible framework provided by crewAI. Our goal is to enable your agents to collaborate effectively on complex tasks, maximizing their collective intelligence and capabilities.
+
+## Installation
+
+Ensure you have Python >=3.10 <=3.13 installed on your system. This project uses [Poetry](https://python-poetry.org/) for dependency management and package handling, offering a seamless setup and execution experience.
+
+First, if you haven't already, install Poetry:
+
+```bash
+pip install poetry
+```
+
+Next, navigate to your project directory and install the dependencies:
+
+1. First lock the dependencies and then install them:
+
+```bash
+crewai install
+```
+
+### Customizing
+
+**Add your `OPENAI_API_KEY` into the `.env` file**
+
+- Modify `src/{{folder_name}}/config/agents.yaml` to define your agents
+- Modify `src/{{folder_name}}/config/tasks.yaml` to define your tasks
+- Modify `src/{{folder_name}}/crew.py` to add your own logic, tools and specific args
+- Modify `src/{{folder_name}}/main.py` to add custom inputs for your agents and tasks
+
+## Running the Project
+
+To kickstart your crew of AI agents and begin task execution, run this from the root folder of your project:
+
+```bash
+crewai run
+```
+
+This command initializes the {{name}} Crew, assembling the agents and assigning them tasks as defined in your configuration.
+
+This example, unmodified, will run the create a `report.md` file with the output of a research on LLMs in the root folder.
+
+## Understanding Your Crew
+
+The {{name}} Crew is composed of multiple AI agents, each with unique roles, goals, and tools. These agents collaborate on a series of tasks, defined in `config/tasks.yaml`, leveraging their collective skills to achieve complex objectives. The `config/agents.yaml` file outlines the capabilities and configurations of each agent in your crew.
+
+## Support
+
+For support, questions, or feedback regarding the {{crew_name}} Crew or crewAI.
+
+- Visit our [documentation](https://docs.crewai.com)
+- Reach out to us through our [GitHub repository](https://github.com/joaomdmoura/crewai)
+- [Join our Discord](https://discord.com/invite/X4JWnZnxPb)
+- [Chat with our docs](https://chatg.pt/DWjSBZn)
+
+Let's create wonders together with the power and simplicity of crewAI.

src/crewai/cli/templates/flow/crews/poem_crew/config/agents.yaml

@@ -0,0 +1,11 @@
+poem_writer:
+  role: >
+    CrewAI Poem Writer
+  goal: >
+    Generate a funny, light heartedpoem about how CrewAI 
+    is awesome with a sentence count of {sentence_count}
+  backstory: >
+    You're a creative poet with a talent for capturing the essence of any topic
+    in a beautiful and engaging way. Known for your ability to craft poems that
+    resonate with readers, you bring a unique perspective and artistic flair to
+    every piece you write.

src/crewai/cli/templates/flow/crews/poem_crew/config/tasks.yaml

@@ -0,0 +1,7 @@
+write_poem:
+  description: >
+    Write a poem about how CrewAI is awesome.
+    Ensure the poem is engaging and adheres to the specified sentence count of {sentence_count}.
+  expected_output: >
+    A beautifully crafted poem about CrewAI, with exactly {sentence_count} sentences.
+  agent: poem_writer

src/crewai/cli/templates/flow/crews/poem_crew/poem_crew.py

@@ -0,0 +1,31 @@
+from crewai import Agent, Crew, Process, Task
+from crewai.project import CrewBase, agent, crew, task
+
+@CrewBase
+class PoemCrew():
+	"""Poem Crew"""
+
+	agents_config = 'config/agents.yaml'
+	tasks_config = 'config/tasks.yaml'
+
+	@agent
+	def poem_writer(self) -> Agent:
+		return Agent(
+			config=self.agents_config['poem_writer'],
+		)
+
+	@task
+	def write_poem(self) -> Task:
+		return Task(
+			config=self.tasks_config['write_poem'],
+		)
+
+	@crew
+	def crew(self) -> Crew:
+		"""Creates the Research Crew"""
+		return Crew(
+			agents=self.agents, # Automatically created by the @agent decorator
+			tasks=self.tasks, # Automatically created by the @task decorator
+			process=Process.sequential,
+			verbose=True,
+		)

src/crewai/cli/templates/flow/main.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python
+import asyncio
+from random import randint
+
+from pydantic import BaseModel
+from crewai.flow.flow import Flow, listen, start
+from .crews.poem_crew.poem_crew import PoemCrew
+
+class PoemState(BaseModel):
+    sentence_count: int = 1
+    poem: str = ""
+
+class PoemFlow(Flow[PoemState]):
+    initial_state = PoemState
+
+    @start()
+    def generate_sentence_count(self):
+        print("Generating sentence count")
+        # Generate a number between 1 and 5
+        self.state.sentence_count = randint(1, 5)  
+
+    @listen(generate_sentence_count)
+    def generate_poem(self):
+        print("Generating poem")
+        print(f"State before poem: {self.state}")
+        poem_crew = PoemCrew().crew()
+        result = poem_crew.kickoff(inputs={"sentence_count": self.state.sentence_count})
+        
+        print("Poem generated", result.raw)
+        self.state.poem = result.raw
+        
+        print(f"State after generate_poem: {self.state}")
+
+    @listen(generate_poem)
+    def save_poem(self):
+        print("Saving poem")
+        print(f"State before save_poem: {self.state}")
+        with open("poem.txt", "w") as f:
+            f.write(self.state.poem)
+        print(f"State after save_poem: {self.state}")
+
+async def run():
+    """
+    Run the flow.
+    """
+    poem_flow = PoemFlow()
+    await poem_flow.kickoff()
+
+
+def main():
+    asyncio.run(run())
+
+
+if __name__ == "__main__":
+    main()

src/crewai/cli/templates/flow/pyproject.toml

@@ -0,0 +1,18 @@
+[tool.poetry]
+name = "{{folder_name}}"
+version = "0.1.0"
+description = "{{name}} using crewAI"
+authors = ["Your Name <you@example.com>"]
+
+[tool.poetry.dependencies]
+python = ">=3.10,<=3.13"
+crewai = { extras = ["tools"], version = ">=0.55.2,<1.0.0" }
+asyncio = "*"
+
+[tool.poetry.scripts]
+{{folder_name}} = "{{folder_name}}.main:main"
+run_crew = "{{folder_name}}.main:main"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

src/crewai/cli/templates/flow/tools/custom_tool.py

@@ -0,0 +1,12 @@
+from crewai_tools import BaseTool
+
+
+class MyCustomTool(BaseTool):
+    name: str = "Name of my tool"
+    description: str = (
+        "Clear description for what this tool is useful for, you agent will need this information to use it."
+    )
+
+    def _run(self, argument: str) -> str:
+        # Implementation goes here
+        return "this is an example of a tool output, ignore it and move along."

src/crewai/cli/templates/pipeline/pyproject.toml

@@ -6,7 +6,7 @@ authors = ["Your Name <you@example.com>"]
 
 [tool.poetry.dependencies]
 python = ">=3.10,<=3.13"
-crewai = { extras = ["tools"], version = ">=0.54.0,<1.0.0" }
+crewai = { extras = ["tools"], version = ">=0.55.2,<1.0.0" }
 asyncio = "*"
 
 [tool.poetry.scripts]

src/crewai/cli/templates/pipeline_router/pyproject.toml

@@ -6,7 +6,7 @@ authors = ["Your Name <you@example.com>"]
 
 [tool.poetry.dependencies]
 python = ">=3.10,<=3.13"
-crewai = { extras = ["tools"], version = ">=0.54.0,<1.0.0" }
+crewai = { extras = ["tools"], version = ">=0.55.2,<1.0.0" }
 
 
 [tool.poetry.scripts]

src/crewai/crews/crew_output.py

@@ -41,6 +41,14 @@ class CrewOutput(BaseModel):
             output_dict.update(self.pydantic.model_dump())
         return output_dict
 
+    def __getitem__(self, key):
+        if self.pydantic and hasattr(self.pydantic, key):
+            return getattr(self.pydantic, key)
+        elif self.json_dict and key in self.json_dict:
+            return self.json_dict[key]
+        else:
+            raise KeyError(f"Key '{key}' not found in CrewOutput.")
+
     def __str__(self):
         if self.pydantic:
             return str(self.pydantic)

src/crewai/flow/decorators.py

@@ -0,0 +1 @@
+# TODO:

src/crewai/flow/examples/context.py

@@ -0,0 +1,13 @@
+from crewai.flows import Flow, end_job, start_job  # type: ignore
+
+
+class SimpleFlow(Flow):
+
+    @start_job()
+    async def research(self, topic: str) -> str:
+        print(f"Researching {topic}...")
+        return f"Full report on {topic}..."
+
+    @end_job("research")
+    async def write_post(self, report: str) -> str:
+        return f"Post written: {report}"

src/crewai/flow/examples/longer.py

@@ -0,0 +1,17 @@
+from crewai.flows import Flow, end_job, job, start_job  # type: ignore
+
+
+class LongerFlow(Flow):
+
+    @start_job()
+    async def research(self, topic: str) -> str:
+        print(f"Researching {topic}...")
+        return f"Full report on {topic}..."
+
+    @job("research")
+    async def edit_report(self, report: str) -> str:
+        return f"Edited report: {report}"
+
+    @end_job("edit_report")
+    async def write_post(self, report: str) -> str:
+        return f"Post written: {report}"

src/crewai/flow/examples/router.py

@@ -0,0 +1,22 @@
+from typing import Tuple
+
+from crewai.flows import Flow, end_job, router, start_job  # type: ignore
+
+
+class RouterFlow(Flow):
+
+    @start_job()
+    @router()
+    async def classify_email(self, report: str) -> Tuple[str, str]:
+        if "urgent" in report:
+            return "urgent", report
+
+        return "normal", report
+
+    @end_job("urgent")
+    async def write_urgent_email(self, report: str) -> str:
+        return f"Urgent Email Response: {report}"
+
+    @end_job("normal")
+    async def write_normal_email(self, report: str) -> str:
+        return f"Normal Email Response: {report}"

src/crewai/flow/examples/simple.py

@@ -0,0 +1,13 @@
+from crewai.flows import Flow, end_job, start_job  # type: ignore
+
+
+class SimpleFlow(Flow):
+
+    @start_job()
+    async def research(self, topic: str) -> str:
+        print(f"Researching {topic}...")
+        return f"Full report on {topic}..."
+
+    @end_job("research")
+    async def write_post(self, report: str) -> str:
+        return f"Post written: {report}"

src/crewai/flow/examples/simple_with_crew.py

@@ -0,0 +1,19 @@
+from crewai.flows import Flow, end_job, job, start_job  # type: ignore
+
+
+class SimpleFlow(Flow):
+
+    @start_job()
+    async def research_crew(self, topic: str) -> str:
+        result = research_crew.kickoff(inputs={topic: topic})
+        return result.raw
+
+    @job("research_crew")
+    async def create_x_post(self, research: str) -> str:
+        result = x_post_crew.kickoff(inputs={research: research})
+        return result.raw
+
+    @end_job("research")
+    async def post_to_x(self, post: str) -> None:
+        # TODO: Post to X
+        return None

src/crewai/flow/flow.py

@@ -0,0 +1,234 @@
+import asyncio
+import inspect
+from typing import Any, Callable, Dict, Generic, List, Set, Type, TypeVar, Union
+
+from pydantic import BaseModel
+
+# TODO: Allow people to pass results from one method to another and not just state
+
+T = TypeVar("T", bound=Union[BaseModel, Dict[str, Any]])
+
+
+def start():
+    def decorator(func):
+        print(f"[start decorator] Decorating start method: {func.__name__}")
+        func.__is_start_method__ = True
+        return func
+
+    return decorator
+
+
+def listen(condition):
+    def decorator(func):
+        print(
+            f"[listen decorator] Decorating listener: {func.__name__} with condition: {condition}"
+        )
+        if isinstance(condition, str):
+            func.__trigger_methods__ = [condition]
+            func.__condition_type__ = "OR"
+            print(
+                f"[listen decorator] Set __trigger_methods__ for {func.__name__}: [{condition}] with mode: OR"
+            )
+        elif (
+            isinstance(condition, dict)
+            and "type" in condition
+            and "methods" in condition
+        ):
+            func.__trigger_methods__ = condition["methods"]
+            func.__condition_type__ = condition["type"]
+            print(
+                f"[listen decorator] Set __trigger_methods__ for {func.__name__}: {func.__trigger_methods__} with mode: {func.__condition_type__}"
+            )
+        elif callable(condition) and hasattr(condition, "__name__"):
+            func.__trigger_methods__ = [condition.__name__]
+            func.__condition_type__ = "OR"
+            print(
+                f"[listen decorator] Set __trigger_methods__ for {func.__name__}: [{condition.__name__}] with mode: OR"
+            )
+        else:
+            raise ValueError(
+                "Condition must be a method, string, or a result of or_() or and_()"
+            )
+        return func
+
+    return decorator
+
+
+def or_(*conditions):
+    methods = []
+    for condition in conditions:
+        if isinstance(condition, dict) and "methods" in condition:
+            methods.extend(condition["methods"])
+        elif callable(condition) and hasattr(condition, "__name__"):
+            methods.append(condition.__name__)
+        elif isinstance(condition, str):
+            methods.append(condition)
+        else:
+            raise ValueError("Invalid condition in or_()")
+    return {"type": "OR", "methods": methods}
+
+
+def and_(*conditions):
+    methods = []
+    for condition in conditions:
+        if isinstance(condition, dict) and "methods" in condition:
+            methods.extend(condition["methods"])
+        elif callable(condition) and hasattr(condition, "__name__"):
+            methods.append(condition.__name__)
+        elif isinstance(condition, str):
+            methods.append(condition)
+        else:
+            raise ValueError("Invalid condition in and_()")
+    return {"type": "AND", "methods": methods}
+
+
+class FlowMeta(type):
+    def __new__(mcs, name, bases, dct):
+        cls = super().__new__(mcs, name, bases, dct)
+
+        start_methods = []
+        listeners = {}
+
+        print(f"[FlowMeta] Processing class: {name}")
+        for attr_name, attr_value in dct.items():
+            print(f"[FlowMeta] Checking attribute: {attr_name}")
+            if hasattr(attr_value, "__is_start_method__"):
+                print(f"[FlowMeta] Found start method: {attr_name}")
+                start_methods.append(attr_name)
+            if hasattr(attr_value, "__trigger_methods__"):
+                methods = attr_value.__trigger_methods__
+                condition_type = getattr(attr_value, "__condition_type__", "OR")
+                print(f"[FlowMeta] Conditions for {attr_name}:", methods)
+                listeners[attr_name] = (condition_type, methods)
+
+        setattr(cls, "_start_methods", start_methods)
+        setattr(cls, "_listeners", listeners)
+
+        print("[FlowMeta] ALL LISTENERS:", listeners)
+        print("[FlowMeta] START METHODS:", start_methods)
+
+        return cls
+
+
+class Flow(Generic[T], metaclass=FlowMeta):
+    _start_methods: List[str] = []
+    _listeners: Dict[str, tuple[str, List[str]]] = {}
+    initial_state: Union[Type[T], T, None] = None
+
+    def __class_getitem__(cls, item):
+        print(f"[Flow.__class_getitem__] Getting initial state type: {item}")
+        class _FlowGeneric(cls):
+            _initial_state_T = item
+        return _FlowGeneric
+
+    def __init__(self):
+        print("[Flow.__init__] Initializing Flow")
+        self._methods: Dict[str, Callable] = {}
+        self._state = self._create_initial_state()
+        self._completed_methods: Set[str] = set()
+        self._pending_and_listeners: Dict[str, Set[str]] = {}
+
+        for method_name in dir(self):
+            if callable(getattr(self, method_name)) and not method_name.startswith(
+                "__"
+            ):
+                print(f"[Flow.__init__] Adding method: {method_name}")
+                self._methods[method_name] = getattr(self, method_name)
+
+        print("[Flow.__init__] All methods:", self._methods.keys())
+        print("[Flow.__init__] Listeners:", self._listeners)
+
+    def _create_initial_state(self) -> T:
+        print("[Flow._create_initial_state] Creating initial state")
+        if self.initial_state is None and hasattr(self, "_initial_state_T"):
+            return self._initial_state_T()
+        elif self.initial_state is None:
+            return {}  # type: ignore
+        elif isinstance(self.initial_state, type):
+            return self.initial_state()
+        else:
+            return self.initial_state
+
+    @property
+    def state(self) -> T:
+        return self._state
+
+    async def kickoff(self):
+        print("[Flow.kickoff] Starting kickoff")
+        if not self._start_methods:
+            raise ValueError("No start method defined")
+
+        for start_method in self._start_methods:
+            print(f"[Flow.kickoff] Executing start method: {start_method}")
+            result = await self._execute_method(self._methods[start_method])
+            print(
+                f"[Flow.kickoff] Start method {start_method} completed. Executing listeners."
+            )
+            await self._execute_listeners(start_method, result)
+
+    async def _execute_method(self, method: Callable, *args, **kwargs):
+        print(f"[Flow._execute_method] Executing method: {method.__name__}")
+        if inspect.iscoroutinefunction(method):
+            return await method(*args, **kwargs)
+        else:
+            return method(*args, **kwargs)
+
+    async def _execute_listeners(self, trigger_method: str, result: Any):
+        print(
+            f"[Flow._execute_listeners] Executing listeners for trigger method: {trigger_method}"
+        )
+        listener_tasks = []
+        for listener, (condition_type, methods) in self._listeners.items():
+            print(
+                f"[Flow._execute_listeners] Checking listener: {listener}, condition: {condition_type}, methods: {methods}"
+            )
+            if condition_type == "OR":
+                if trigger_method in methods:
+                    print(
+                        f"[Flow._execute_listeners] TRIGGERING METHOD: {listener} due to trigger: {trigger_method}"
+                    )
+                    listener_tasks.append(
+                        self._execute_single_listener(listener, result)
+                    )
+            elif condition_type == "AND":
+                if listener not in self._pending_and_listeners:
+                    self._pending_and_listeners[listener] = set()
+                self._pending_and_listeners[listener].add(trigger_method)
+                if set(methods) == self._pending_and_listeners[listener]:
+                    print(
+                        f"[Flow._execute_listeners] All conditions met for listener: {listener}. Executing."
+                    )
+                    listener_tasks.append(
+                        self._execute_single_listener(listener, result)
+                    )
+                    del self._pending_and_listeners[listener]
+
+        # Run all listener tasks concurrently and wait for them to complete
+        print(
+            f"[Flow._execute_listeners] Executing {len(listener_tasks)} listener tasks"
+        )
+        await asyncio.gather(*listener_tasks)
+
+    async def _execute_single_listener(self, listener: str, result: Any):
+        print(f"[Flow._execute_single_listener] Executing listener: {listener}")
+        try:
+            method = self._methods[listener]
+            sig = inspect.signature(method)
+            if len(sig.parameters) > 1:  # More than just 'self'
+                print(
+                    f"[Flow._execute_single_listener] Executing {listener} with result"
+                )
+                listener_result = await self._execute_method(method, result)
+            else:
+                print(
+                    f"[Flow._execute_single_listener] Executing {listener} without result"
+                )
+                listener_result = await self._execute_method(method)
+            print(
+                f"[Flow._execute_single_listener] {listener} completed, executing its listeners"
+            )
+            await self._execute_listeners(listener, listener_result)
+        except Exception as e:
+            print(
+                f"[Flow._execute_single_listener] Error in method {listener}: {str(e)}"
+            )

src/crewai/flow/structured_test_flow.py

@@ -0,0 +1,46 @@
+import asyncio
+
+from crewai.flow.flow import Flow, listen, start
+from pydantic import BaseModel
+
+
+class ExampleState(BaseModel):
+    counter: int = 0
+    message: str = ""
+
+
+class StructuredExampleFlow(Flow[ExampleState]):
+
+    @start()
+    async def start_method(self):
+        print("Starting the structured flow")
+        print(f"State in start_method: {self.state}")
+        self.state.message = "Hello from structured flow"
+        print(f"State after start_method: {self.state}")
+        return "Start result"
+
+    @listen(start_method)
+    async def second_method(self, result):
+        print(f"Second method, received: {result}")
+        print(f"State before increment: {self.state}")
+        self.state.counter += 1
+        self.state.message += " - updated"
+        print(f"State after second_method: {self.state}")
+        return "Second result"
+
+    @listen(start_method)
+    async def third_method(self, result):
+        print(f"Third method, received: {result}")
+        print(f"State before increment: {self.state}")
+        self.state.counter += 1
+        self.state.message += " - updated"
+        print(f"State after third_method: {self.state}")
+        return "Third result"
+
+
+async def main():
+    flow = StructuredExampleFlow()
+    await flow.kickoff()
+
+
+asyncio.run(main())

src/crewai/flow/structured_test_flow_and.py

@@ -0,0 +1,42 @@
+import asyncio
+
+from crewai.flow.flow import Flow, listen, start
+from pydantic import BaseModel
+
+
+class ExampleState(BaseModel):
+    counter: int = 0
+    message: str = ""
+
+
+class StructuredExampleFlow(Flow):
+    initial_state = ExampleState
+
+    @start()
+    async def start_method(self):
+        print("Starting the structured flow")
+        print(f"State in start_method: {self.state}")
+        self.state.message = "Hello from structured flow"
+        print(f"State after start_method: {self.state}")
+        return "Start result"
+
+    @listen(start_method)
+    async def second_method(self):
+        print(f"State before increment: {self.state}")
+        self.state.counter += 1
+        self.state.message += " - updated"
+        print(f"State after second_method: {self.state}")
+        return "Second result"
+
+    @listen(start_method and second_method)
+    async def logger(self):
+        print("AND METHOD RUNNING")
+        print("CURRENT STATE FROM OR: ", self.state)
+
+
+async def main():
+    flow = StructuredExampleFlow()
+    await flow.kickoff()
+
+
+asyncio.run(main())

src/crewai/flow/structured_test_flow_or.py

@@ -0,0 +1,47 @@
+import asyncio
+
+from crewai.flow.flow import Flow, and_, listen, or_, start
+from pydantic import BaseModel
+
+
+class ExampleState(BaseModel):
+    counter: int = 0
+    message: str = ""
+
+
+class StructuredExampleFlow(Flow[ExampleState]):
+    initial_state = ExampleState
+
+    @start()
+    async def start_method(self):
+        print("Starting the structured flow")
+        print(f"State in start_method: {self.state}")
+        self.state.message = "Hello from structured flow"
+        print(f"State after start_method: {self.state}")
+        return "Start result"
+
+    @listen(start_method)
+    async def second_method(self):
+        print(f"State before increment: {self.state}")
+        self.state.counter += 1
+        self.state.message += " - updated"
+        print(f"State after second_method: {self.state}")
+        return "Second result"
+
+    @listen(or_(start_method, second_method))
+    async def logger(self):
+        print("LOGGER METHOD RUNNING")
+        print("CURRENT STATE FROM LOGGER: ", self.state)
+
+    @listen(and_(start_method, second_method))
+    async def and_logger(self):
+        print("AND LOGGER METHOD RUNNING")
+        print("CURRENT STATE FROM AND LOGGER: ", self.state)
+
+
+async def main():
+    flow = StructuredExampleFlow()
+    await flow.kickoff()
+
+
+asyncio.run(main())

src/crewai/flow/unstructured_test_flow.py

@@ -0,0 +1,33 @@
+import asyncio
+
+from crewai.flow.flow import Flow, listen, start
+
+
+class FlexibleExampleFlow(Flow):
+    @start()
+    def start_method(self):
+        print("Starting the flexible flow")
+        self.state["counter"] = 1
+        return "Start result"
+
+    @listen(start_method)
+    def second_method(self, result):
+        print(f"Second method, received: {result}")
+        self.state["counter"] += 1
+        self.state["message"] = "Hello from flexible flow"
+        return "Second result"
+
+    @listen(second_method)
+    def third_method(self, result):
+        print(f"Third method, received: {result}")
+        print(f"Final counter value: {self.state["counter"]}")
+        print(f"Final message: {self.state["message"]}")
+        return "Third result"
+
+
+async def main():
+    flow = FlexibleExampleFlow()
+    await flow.kickoff()
+
+
+asyncio.run(main())