Consolidate crewai run and crewai flow kickoff (#6296)

Make `crewai run` the single execution path for crews and flows, with
`crewai flow kickoff` kept as a deprecated compatibility alias.

docs/edge/ar/changelog.mdx

@@ -64,7 +64,7 @@ mode: "wide"
   - تنفيذ أدوات تشغيل تعريف التدفق بدون كود Python
   - دفع التغذية الراجعة البشرية من تعريف التدفق
   - توصيل التكوين والاستمرارية من FlowDefinition إلى وقت التشغيل
-  - إضافة `crewai run --definition` التجريبية للتدفقات
+  - إضافة `crewai run --definition` للتدفقات التصريحية
   - دعم تراجع نشر ZIP وتشغيل مشاريع الطاقم بتنسيق JSON
   - تقديم الطواقم بتنسيق JSON أولاً
 

docs/edge/ar/concepts/flows.mdx

@@ -959,7 +959,7 @@ source .venv/bin/activate
 بعد تفعيل البيئة الافتراضية، يمكنك تشغيل التدفق بتنفيذ أحد الأوامر التالية:
 
 ```bash
-crewai flow kickoff
+crewai run
 ```
 
 أو
@@ -1160,10 +1160,4 @@ crewai run
 
 يكتشف هذا الأمر تلقائيًا ما إذا كان مشروعك تدفقًا (بناءً على إعداد `type = "flow"` في pyproject.toml الخاص بك) ويشغّله وفقًا لذلك. هذه هي الطريقة الموصى بها لتشغيل التدفقات من سطر الأوامر.
 
-للتوافق مع الإصدارات السابقة، يمكنك أيضًا استخدام:
-
-```shell
-crewai flow kickoff
-```
-
-ومع ذلك، فإن أمر `crewai run` هو الطريقة المفضلة الآن لأنه يعمل لكل من فرق Crew والتدفقات.
+أمر `crewai flow kickoff` القديم deprecated. استخدم `crewai run` لكل من فرق Crew والتدفقات.

docs/edge/ar/guides/flows/first-flow.mdx

@@ -172,7 +172,7 @@ crewai install
 ## الخطوة 8: تشغيل Flow
 
 ```bash
-crewai flow kickoff
+crewai run
 ```
 
 عند تشغيل هذا الأمر، ستشاهد Flow يعمل:

docs/edge/en/changelog.mdx

@@ -64,7 +64,7 @@ mode: "wide"
   - Implement Flow definition run tools without Python code
   - Drive human feedback from the flow definition
   - Wire config and persistence from FlowDefinition into the runtime
-  - Add experimental `crewai run --definition` for flows
+  - Add `crewai run --definition` for declarative flows
   - Support ZIP deployment fallback and JSON crew project env runs
   - Introduce JSON first crews
 

docs/edge/en/concepts/flows.mdx

@@ -956,13 +956,13 @@ Once all of the dependencies are installed, you need to activate the virtual env
 source .venv/bin/activate
 ```
 
-After activating the virtual environment, you can run the flow by executing one of the following commands:
+After activating the virtual environment, you can run the flow with the CrewAI CLI:
 
 ```bash
-crewai flow kickoff
+crewai run
 ```
 
-or
+You can also run the project script directly:
 
 ```bash
 uv run kickoff
@@ -1160,10 +1160,4 @@ crewai run
 
 This command automatically detects if your project is a flow (based on the `type = "flow"` setting in your pyproject.toml) and runs it accordingly. This is the recommended way to run flows from the command line.
 
-For backward compatibility, you can also use:
-
-```shell
-crewai flow kickoff
-```
-
-However, the `crewai run` command is now the preferred method as it works for both crews and flows.
+The legacy `crewai flow kickoff` command is deprecated. Use `crewai run` for both crews and flows.

docs/edge/en/guides/flows/first-flow.mdx

@@ -395,7 +395,7 @@ crewai install
 Now it's time to see your flow in action! Run it using the CrewAI CLI:
 
 ```bash
-crewai flow kickoff
+crewai run
 ```
 
 When you run this command, you'll see your flow spring to life:

docs/edge/ko/changelog.mdx

@@ -64,7 +64,7 @@ mode: "wide"
   - Python 코드 없이 Flow 정의 실행 도구 구현
   - Flow 정의에서 인간 피드백 유도
   - FlowDefinition의 구성 및 지속성을 런타임에 연결
-  - 흐름을 위한 실험적 `crewai run --definition` 추가
+  - 선언적 흐름을 위한 `crewai run --definition` 추가
   - ZIP 배포 대체 및 JSON 크루 프로젝트 환경 실행 지원
   - JSON 우선 크루 도입
 

docs/edge/ko/concepts/flows.mdx

@@ -951,7 +951,7 @@ source .venv/bin/activate
 가상 환경을 활성화한 후, 아래 명령어 중 하나를 실행하여 플로우를 실행할 수 있습니다:
 
 ```bash
-crewai flow kickoff
+crewai run
 ```
 
 또는
@@ -1054,10 +1054,4 @@ crewai run
 
 이 명령어는 프로젝트가 pyproject.toml의 `type = "flow"` 설정을 기반으로 flow인지 자동으로 감지하여 해당 방식으로 실행합니다. 명령줄에서 flow를 실행하는 권장 방법입니다.
 
-하위 호환성을 위해 다음 명령어도 사용할 수 있습니다:
-
-```shell
-crewai flow kickoff
-```
-
-하지만 `crewai run` 명령어가 이제 crew와 flow 모두에 작동하므로 더욱 선호되는 방법입니다.
+레거시 `crewai flow kickoff` 명령어는 deprecated되었습니다. crew와 flow 모두 `crewai run`을 사용하세요.

docs/edge/ko/guides/flows/first-flow.mdx

@@ -393,7 +393,7 @@ crewai install
 이제 여러분의 flow가 실제로 작동하는 모습을 볼 차례입니다! CrewAI CLI를 사용하여 flow를 실행하세요:
 
 ```bash
-crewai flow kickoff
+crewai run
 ```
 
 이 명령어를 실행하면 flow가 다음과 같이 작동하는 것을 확인할 수 있습니다:

docs/edge/pt-BR/changelog.mdx

@@ -64,7 +64,7 @@ mode: "wide"
   - Implementar ferramentas de execução de definição de fluxo sem código Python
   - Conduzir feedback humano a partir da definição de fluxo
   - Conectar configuração e persistência do FlowDefinition ao tempo de execução
-  - Adicionar `crewai run --definition` experimental para fluxos
+  - Adicionar `crewai run --definition` para fluxos declarativos
   - Suportar fallback de implantação ZIP e execuções de projeto de equipe em JSON
   - Introduzir equipes em JSON primeiro
 

docs/edge/pt-BR/concepts/flows.mdx

@@ -948,7 +948,7 @@ source .venv/bin/activate
 Com o ambiente ativado, execute o flow usando um dos comandos:
 
 ```bash
-crewai flow kickoff
+crewai run
 ```
 
 ou
@@ -1052,10 +1052,4 @@ crewai run
 
 O comando detecta automaticamente se seu projeto é um flow (com base na configuração `type = "flow"` no pyproject.toml) e executa conforme o esperado. Esse é o método recomendado para executar flows pelo terminal.
 
-Por compatibilidade retroativa, também é possível usar:
-
-```shell
-crewai flow kickoff
-```
-
-No entanto, o comando `crewai run` é agora o preferido, pois funciona tanto para crews quanto para flows.
+O comando legado `crewai flow kickoff` está deprecated. Use `crewai run` para crews e flows.

docs/edge/pt-BR/guides/flows/first-flow.mdx

@@ -393,7 +393,7 @@ crewai install
 Agora é hora de ver seu flow em ação! Execute-o usando a CLI do CrewAI:
 
 ```bash
-crewai flow kickoff
+crewai run
 ```
 
 Quando você rodar esse comando, verá seu flow ganhando vida:

lib/cli/src/crewai_cli/cli.py

@@ -40,14 +40,6 @@ def replay_task_command(*args: Any, **kwargs: Any) -> Any:
     return _replay_task_command(*args, **kwargs)
 
 
-def run_declarative_flow(*args: Any, **kwargs: Any) -> Any:
-    from crewai_cli.run_declarative_flow import (
-        run_declarative_flow as _run_declarative_flow,
-    )
-
-    return _run_declarative_flow(*args, **kwargs)
-
-
 def run_crew(*args: Any, **kwargs: Any) -> Any:
     from crewai_cli.run_crew import run_crew as _run_crew
 
@@ -476,7 +468,7 @@ def memory(
     type=str,
     default=None,
     help=(
-        "Path to a trained-agents pickle (produced by `crewai train -f`). "
+        "Crew-only: path to a trained-agents pickle (produced by `crewai train -f`). "
         "When set, agents load suggestions from this file instead of the "
         "default trained_agents_data.pkl. Equivalent to setting "
         "CREWAI_TRAINED_AGENTS_FILE."
@@ -520,13 +512,13 @@ def install(context: click.Context) -> None:
     "--definition",
     type=str,
     default=None,
-    help="Experimental: path to a declarative Flow YAML/JSON file.",
+    help="Flow-only: path to a declarative flow definition.",
 )
 @click.option(
     "--inputs",
     type=str,
     default=None,
-    help='Experimental: JSON object passed to flow.kickoff(), e.g. \'{"topic":"AI"}\'.',
+    help='Flow-only: JSON object passed to the declarative flow, e.g. \'{"topic":"AI"}\'.',
 )
 def run(
     trained_agents_file: str | None,
@@ -536,16 +528,14 @@ def run(
     """Run the Crew or Flow."""
     if inputs is not None and definition is None:
         raise click.UsageError("--inputs requires --definition")
+    if trained_agents_file is not None and definition is not None:
+        raise click.UsageError("--filename can only be used when running crews")
 
-    if definition is not None:
-        click.secho(
-            "Warning: `crewai run --definition` is experimental and may change without notice.",
-            fg="yellow",
-        )
-        run_declarative_flow(definition=definition, inputs=inputs)
-        return
-
-    run_crew(trained_agents_file=trained_agents_file)
+    run_crew(
+        trained_agents_file=trained_agents_file,
+        definition=definition,
+        inputs=inputs,
+    )
 
 
 @crewai.command()
@@ -797,13 +787,10 @@ def flow() -> None:
     """Flow related commands."""
 
 
-@flow.command(name="kickoff")
+@flow.command(name="kickoff", deprecated=True)
 def flow_run() -> None:
     """Kickoff the Flow."""
-    from crewai_cli.kickoff_flow import kickoff_flow
-
-    click.echo("Running the Flow")
-    kickoff_flow()
+    run_crew(trained_agents_file=None, definition=None, inputs=None)
 
 
 @flow.command(name="plot")

lib/cli/src/crewai_cli/kickoff_flow.py

@@ -1,31 +0,0 @@
-import subprocess
-
-import click
-
-
-def kickoff_flow() -> None:
-    """
-    Kickoff the flow from declarative config or the Python UV entrypoint.
-    """
-    from crewai_cli.run_declarative_flow import (
-        configured_project_declarative_flow,
-        run_declarative_flow_in_project_env,
-    )
-
-    if definition := configured_project_declarative_flow():
-        run_declarative_flow_in_project_env(definition=definition)
-    else:
-        command = ["uv", "run", "kickoff"]
-
-        try:
-            subprocess.run(  # noqa: S603
-                command, capture_output=False, text=True, check=True
-            )
-
-        except subprocess.CalledProcessError as e:
-            click.echo(f"An error occurred while running the flow: {e}", err=True)
-            raise SystemExit(1) from e
-
-        except Exception as e:
-            click.echo(f"An unexpected error occurred: {e}", err=True)
-            raise SystemExit(1) from e

lib/cli/src/crewai_cli/run_crew.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 from collections.abc import Callable
 from contextlib import AbstractContextManager, nullcontext
-from enum import Enum
 import os
 from pathlib import Path
 import re
@@ -27,11 +26,6 @@ if TYPE_CHECKING:
     from crewai_cli.crew_run_tui import CrewRunApp
 
 
-class CrewType(Enum):
-    STANDARD = "standard"
-    FLOW = "flow"
-
-
 # Must accept the same names as the kickoff interpolation pattern in
 # crewai.utilities.string_utils (_VARIABLE_PATTERN), including hyphens —
 # otherwise placeholders are interpolated at runtime but never prompted for.
@@ -537,7 +531,11 @@ def _print_post_tui_summary(app: CrewRunApp) -> None:
         )
 
 
-def run_crew(trained_agents_file: str | None = None) -> None:
+def run_crew(
+    trained_agents_file: str | None = None,
+    definition: str | None = None,
+    inputs: str | None = None,
+) -> None:
     """Run the crew or flow.
 
     Args:
@@ -545,15 +543,90 @@ def run_crew(trained_agents_file: str | None = None) -> None:
             by ``crewai train -f``. When set, exported as
             ``CREWAI_TRAINED_AGENTS_FILE`` so agents load suggestions from this
             file instead of the default ``trained_agents_data.pkl``.
+        definition: Optional path to a declarative Flow definition.
+        inputs: Optional JSON object passed to a declarative Flow.
     """
-    # JSON crew projects take precedence
+    if inputs is not None and definition is None:
+        raise click.UsageError("--inputs requires --definition")
+
+    if definition is not None:
+        _run_explicit_declarative_flow(
+            definition=definition,
+            inputs=inputs,
+            trained_agents_file=trained_agents_file,
+        )
+        return
+
     if _has_json_crew():
         _run_json_crew_in_project_env(trained_agents_file=trained_agents_file)
         return
 
+    pyproject_data = read_toml()
+    _warn_if_old_poetry_project(pyproject_data)
+    project_type = _get_project_type(pyproject_data)
+
+    if project_type == "flow":
+        _run_flow_project(
+            pyproject_data=pyproject_data,
+            trained_agents_file=trained_agents_file,
+        )
+        return
+
+    _run_classic_crew_project(
+        pyproject_data=pyproject_data,
+        trained_agents_file=trained_agents_file,
+    )
+
+
+def _run_explicit_declarative_flow(
+    definition: str, inputs: str | None, trained_agents_file: str | None
+) -> None:
+    if trained_agents_file is not None:
+        raise click.UsageError("--filename can only be used when running crews")
+
+    from crewai_cli.run_declarative_flow import run_declarative_flow
+
+    run_declarative_flow(definition=definition, inputs=inputs)
+
+
+def _run_flow_project(
+    pyproject_data: dict[str, Any], trained_agents_file: str | None
+) -> None:
+    if trained_agents_file is not None:
+        raise click.UsageError("--filename can only be used when running crews")
+
+    click.echo("Running the Flow")
+    from crewai_cli.run_declarative_flow import (
+        configured_project_declarative_flow,
+        run_declarative_flow_in_project_env,
+    )
+
+    if definition := configured_project_declarative_flow(pyproject_data):
+        run_declarative_flow_in_project_env(definition=definition)
+        return
+
+    _execute_uv_script("kickoff", entity_type="flow")
+
+
+def _run_classic_crew_project(
+    pyproject_data: dict[str, Any], trained_agents_file: str | None
+) -> None:
+    click.echo("Running the Crew")
+    _execute_uv_script(
+        "run_crew",
+        entity_type="crew",
+        trained_agents_file=trained_agents_file,
+    )
+
+
+def _get_project_type(pyproject_data: dict[str, Any]) -> str | None:
+    project_type = pyproject_data.get("tool", {}).get("crewai", {}).get("type")
+    return project_type if isinstance(project_type, str) else None
+
+
+def _warn_if_old_poetry_project(pyproject_data: dict[str, Any]) -> None:
     crewai_version = get_crewai_version()
     min_required_version = "0.71.0"
-    pyproject_data = read_toml()
 
     if pyproject_data.get("tool", {}).get("poetry") and (
         version.parse(crewai_version) < version.parse(min_required_version)
@@ -564,25 +637,22 @@ def run_crew(trained_agents_file: str | None = None) -> None:
             fg="red",
         )
 
-    is_flow = pyproject_data.get("tool", {}).get("crewai", {}).get("type") == "flow"
-    crew_type = CrewType.FLOW if is_flow else CrewType.STANDARD
 
-    click.echo(f"Running the {'Flow' if is_flow else 'Crew'}")
-
-    execute_command(crew_type, trained_agents_file=trained_agents_file)
-
-
-def execute_command(
-    crew_type: CrewType, trained_agents_file: str | None = None
+def _execute_uv_script(
+    script_name: str,
+    *,
+    entity_type: str,
+    trained_agents_file: str | None = None,
 ) -> None:
-    """Execute the appropriate command based on crew type.
+    """Execute a project script through uv.
 
     Args:
-        crew_type: The type of crew to run.
+        script_name: The project script to run.
+        entity_type: The user-facing entity being run.
         trained_agents_file: Optional trained-agents pickle path forwarded to
             the subprocess via the ``CREWAI_TRAINED_AGENTS_FILE`` env var.
     """
-    command = ["uv", "run", "kickoff" if crew_type == CrewType.FLOW else "run_crew"]
+    command = ["uv", "run", script_name]
 
     env = build_env_with_all_tool_credentials()
     if trained_agents_file:
@@ -592,21 +662,20 @@ def execute_command(
         subprocess.run(command, capture_output=False, text=True, check=True, env=env)  # noqa: S603
 
     except subprocess.CalledProcessError as e:
-        handle_error(e, crew_type)
+        _handle_run_error(e, entity_type)
 
     except Exception as e:
         click.echo(f"An unexpected error occurred: {e}", err=True)
 
 
-def handle_error(error: subprocess.CalledProcessError, crew_type: CrewType) -> None:
+def _handle_run_error(error: subprocess.CalledProcessError, entity_type: str) -> None:
     """
     Handle subprocess errors with appropriate messaging.
 
     Args:
         error: The subprocess error that occurred
-        crew_type: The type of crew that was being run
+        entity_type: The type of entity that was being run
     """
-    entity_type = "flow" if crew_type == CrewType.FLOW else "crew"
     click.echo(f"An error occurred while running the {entity_type}: {error}", err=True)
 
     if error.output:

lib/cli/src/crewai_cli/run_declarative_flow.py

@@ -21,7 +21,7 @@ def run_declarative_flow_in_project_env(
     if inputs is not None:
         raise click.UsageError("--inputs is only supported with --definition")
 
-    _execute_declarative_flow_command(["uv", "run", "crewai", "flow", "kickoff"])
+    _execute_declarative_flow_command(["uv", "run", "crewai", "run"])
 
 
 def plot_declarative_flow_in_project_env(definition: str) -> None:
@@ -34,7 +34,7 @@ def plot_declarative_flow_in_project_env(definition: str) -> None:
 
 
 def run_declarative_flow(definition: str, inputs: str | None = None) -> None:
-    """Run a declarative flow from a YAML/JSON file path."""
+    """Run a declarative flow from a definition path."""
     parsed_inputs = _parse_inputs(inputs)
 
     try:
@@ -50,7 +50,7 @@ def run_declarative_flow(definition: str, inputs: str | None = None) -> None:
 
 
 def plot_declarative_flow(definition: str) -> None:
-    """Plot a declarative flow from a YAML/JSON file path."""
+    """Plot a declarative flow from a definition path."""
     try:
         flow = load_declarative_flow(definition)
         flow.plot()
@@ -62,7 +62,7 @@ def plot_declarative_flow(definition: str) -> None:
 
 
 def load_declarative_flow(definition: str) -> Any:
-    """Load a declarative Flow instance from a YAML/JSON file path."""
+    """Load a declarative Flow instance from a definition path."""
     try:
         from crewai.flow.flow import Flow
         from crewai.flow.flow_definition import FlowDefinition

lib/cli/src/crewai_cli/templates/AGENTS.md

@@ -62,7 +62,7 @@ crewai create flow <name> --skip_provider  # New flow project
 
 # Running
 crewai run                  # Run crew or flow (auto-detects from pyproject.toml)
-crewai flow kickoff         # Legacy flow execution
+crewai flow kickoff         # Deprecated compatibility alias for crewai run
 
 # Testing & training
 crewai test                           # Test crew (default: 2 iterations, gpt-4o-mini)

lib/cli/src/crewai_cli/templates/declarative_flow/README.md

@@ -1,6 +1,6 @@
 # {{name}} Flow
 
-This project defines a CrewAI Flow in `src/{{folder_name}}/flow.yaml`.
+This project defines a declarative CrewAI Flow in `src/{{folder_name}}/flow.yaml`.
 
 ## Install
 
@@ -11,7 +11,7 @@ crewai install
 ## Run
 
 ```bash
-crewai flow kickoff
+crewai run
 ```
 
-Edit `src/{{folder_name}}/flow.yaml` to change the flow. Add reusable crews under `src/{{folder_name}}/crews/`, custom Python tools under `src/{{folder_name}}/tools/`, and shared knowledge files under `src/{{folder_name}}/knowledge/`.
+Edit the declarative flow definition at `src/{{folder_name}}/flow.yaml` to change the flow. Add reusable crews under `src/{{folder_name}}/crews/`, custom Python tools under `src/{{folder_name}}/tools/`, and shared knowledge files under `src/{{folder_name}}/knowledge/`.

lib/cli/tests/test_cli.py

@@ -12,6 +12,7 @@ from crewai_cli.cli import (
     deploy_remove,
     deply_status,
     flow_add_crew,
+    flow_run,
     login,
     reset_memories,
     run,
@@ -126,40 +127,72 @@ def test_run_uses_project_runner_by_default(run_crew, runner):
     result = runner.invoke(run)
 
     assert result.exit_code == 0
-    run_crew.assert_called_once_with(trained_agents_file=None)
+    run_crew.assert_called_once_with(
+        trained_agents_file=None,
+        definition=None,
+        inputs=None,
+    )
     assert "experimental" not in result.output.lower()
 
 
-@mock.patch("crewai_cli.cli.run_declarative_flow")
-def test_run_with_definition_uses_definition_runner(run_declarative_flow, runner):
+@mock.patch("crewai_cli.cli.run_crew")
+def test_run_with_definition_uses_project_runner(run_crew, runner):
     result = runner.invoke(
         run,
         ["--definition", "flow.yaml", "--inputs", '{"topic":"AI"}'],
     )
 
     assert result.exit_code == 0
-    assert (
-        "Warning: `crewai run --definition` is experimental and may change without notice."
-        in result.output
-    )
-    run_declarative_flow.assert_called_once_with(
-        definition="flow.yaml", inputs='{"topic":"AI"}'
+    run_crew.assert_called_once_with(
+        trained_agents_file=None,
+        definition="flow.yaml",
+        inputs='{"topic":"AI"}',
     )
 
 
 @mock.patch("crewai_cli.cli.run_crew")
-@mock.patch("crewai_cli.cli.run_declarative_flow")
-def test_run_rejects_inputs_without_definition(
-    run_declarative_flow, run_crew, runner
-):
+def test_run_rejects_inputs_without_definition(run_crew, runner):
     result = runner.invoke(run, ["--inputs", '{"topic":"AI"}'])
 
     assert result.exit_code == 2
     assert "Error: --inputs requires --definition" in result.output
-    run_declarative_flow.assert_not_called()
     run_crew.assert_not_called()
 
 
+@mock.patch("crewai_cli.cli.run_crew")
+def test_run_rejects_filename_with_definition(run_crew, runner):
+    result = runner.invoke(run, ["--definition", "flow.yaml", "--filename", "x.pkl"])
+
+    assert result.exit_code == 2
+    assert "Error: --filename can only be used when running crews" in result.output
+    run_crew.assert_not_called()
+
+
+@mock.patch("crewai_cli.cli.run_crew")
+def test_run_passes_filename_to_project_runner(run_crew, runner):
+    result = runner.invoke(run, ["--filename", "trained.pkl"])
+
+    assert result.exit_code == 0
+    run_crew.assert_called_once_with(
+        trained_agents_file="trained.pkl",
+        definition=None,
+        inputs=None,
+    )
+
+
+@mock.patch("crewai_cli.cli.run_crew")
+def test_flow_kickoff_is_deprecated_and_uses_run_path(run_crew, runner):
+    result = runner.invoke(flow_run)
+
+    assert result.exit_code == 0
+    run_crew.assert_called_once_with(
+        trained_agents_file=None,
+        definition=None,
+        inputs=None,
+    )
+    assert "DeprecationWarning" in result.output
+
+
 @mock.patch("crewai_cli.create_json_crew.create_json_crew")
 def test_create_crew_in_dmn_mode_skips_provider_prompts(create_json_crew, runner):
     result = runner.invoke(create, ["crew", "DMN Crew"], env={"CREWAI_DMN": "True"})

lib/cli/tests/test_create_flow.py

@@ -28,9 +28,7 @@ def test_create_flow_declarative_project_can_run(
     assert (project_root / pyproject["tool"]["crewai"]["definition"]).is_file()
 
     monkeypatch.chdir(project_root)
-    result = CliRunner().invoke(
-        crewai, ["flow", "kickoff"], env={"UV_RUN_RECURSION_DEPTH": "1"}
-    )
+    result = CliRunner().invoke(crewai, ["run"], env={"UV_RUN_RECURSION_DEPTH": "1"})
 
     assert result.exit_code == 0
     assert "Running the Flow" in result.output

lib/cli/tests/test_flow_commands.py

@@ -1,12 +1,12 @@
 from __future__ import annotations
 
-from collections.abc import Callable
 from pathlib import Path
 import subprocess
 
 import pytest
+from click.testing import CliRunner
 
-import crewai_cli.kickoff_flow as kickoff_flow_module
+from crewai_cli.cli import flow_run
 import crewai_cli.plot_flow as plot_flow_module
 
 
@@ -33,18 +33,19 @@ def _write_flow_project(project_root: Path) -> None:
     )
 
 
-def test_kickoff_flow_runs_configured_declarative_definition(
+def test_flow_kickoff_runs_configured_declarative_definition(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
-    capsys: pytest.CaptureFixture[str],
 ) -> None:
     _write_flow_project(tmp_path)
     monkeypatch.chdir(tmp_path)
     monkeypatch.setenv("UV_RUN_RECURSION_DEPTH", "1")
 
-    kickoff_flow_module.kickoff_flow()
+    result = CliRunner().invoke(flow_run)
 
-    assert capsys.readouterr().out == "AI\n"
+    assert result.exit_code == 0
+    assert "DeprecationWarning" in result.output
+    assert "Running the Flow\nAI\n" in result.output
 
 
 def test_plot_flow_runs_configured_declarative_definition(
@@ -57,18 +58,27 @@ def test_plot_flow_runs_configured_declarative_definition(
     plot_flow_module.plot_flow()
 
 
-@pytest.mark.parametrize(
-    ("command", "expected"),
-    [
-        pytest.param(kickoff_flow_module.kickoff_flow, ["uv", "run", "kickoff"]),
-        pytest.param(plot_flow_module.plot_flow, ["uv", "run", "plot"]),
-    ],
-)
-def test_flow_commands_keep_python_entrypoint_without_definition(
+def test_flow_kickoff_delegates_to_run_crew(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    calls = []
+
+    monkeypatch.setattr(
+        "crewai_cli.cli.run_crew",
+        lambda **kwargs: calls.append(kwargs),
+    )
+
+    result = CliRunner().invoke(flow_run)
+
+    assert result.exit_code == 0
+    assert calls == [
+        {"trained_agents_file": None, "definition": None, "inputs": None},
+    ]
+
+
+def test_plot_flow_keeps_python_entrypoint_without_definition(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
-    command: Callable[[], None],
-    expected: list[str],
 ) -> None:
     subprocess_calls = []
 
@@ -79,11 +89,11 @@ def test_flow_commands_keep_python_entrypoint_without_definition(
         lambda command, **kwargs: subprocess_calls.append((command, kwargs)),
     )
 
-    command()
+    plot_flow_module.plot_flow()
 
     assert subprocess_calls == [
         (
-            expected,
+            ["uv", "run", "plot"],
             {"capture_output": False, "text": True, "check": True},
         )
     ]

lib/cli/tests/test_run_crew.py

@@ -568,3 +568,131 @@ def test_has_json_crew_true_without_pyproject(monkeypatch, tmp_path: Path):
     (tmp_path / "crew.jsonc").write_text("{}")
 
     assert run_crew_module._has_json_crew() is True
+
+
+def test_run_crew_rejects_inputs_without_definition():
+    with pytest.raises(click.UsageError) as exc_info:
+        run_crew_module.run_crew(inputs='{"topic":"AI"}')
+
+    assert "--inputs requires --definition" in exc_info.value.message
+
+
+def test_run_crew_rejects_filename_with_explicit_definition():
+    with pytest.raises(click.UsageError) as exc_info:
+        run_crew_module.run_crew(
+            trained_agents_file="trained.pkl",
+            definition="flow.yaml",
+        )
+
+    assert "--filename can only be used when running crews" in exc_info.value.message
+
+
+def test_run_crew_runs_explicit_declarative_definition(monkeypatch, capsys):
+    calls = []
+
+    def fake_run_declarative_flow(definition: str, inputs: str | None = None):
+        calls.append((definition, inputs))
+
+    monkeypatch.setattr(
+        "crewai_cli.run_declarative_flow.run_declarative_flow",
+        fake_run_declarative_flow,
+    )
+
+    run_crew_module.run_crew(definition="flow.yaml", inputs='{"topic":"AI"}')
+
+    captured = capsys.readouterr()
+    assert "experimental" not in captured.out.lower()
+    assert calls == [("flow.yaml", '{"topic":"AI"}')]
+
+
+def test_run_crew_runs_classic_crew_project(monkeypatch, capsys):
+    calls = []
+
+    monkeypatch.setattr(run_crew_module, "_has_json_crew", lambda: False)
+    monkeypatch.setattr(
+        run_crew_module,
+        "read_toml",
+        lambda: {"tool": {"crewai": {"type": "crew"}}},
+    )
+    monkeypatch.setattr(
+        run_crew_module,
+        "_execute_uv_script",
+        lambda script_name, **kwargs: calls.append((script_name, kwargs)),
+    )
+
+    run_crew_module.run_crew(trained_agents_file="trained.pkl")
+
+    assert capsys.readouterr().out == "Running the Crew\n"
+    assert calls == [
+        (
+            "run_crew",
+            {"entity_type": "crew", "trained_agents_file": "trained.pkl"},
+        )
+    ]
+
+
+def test_run_crew_runs_python_flow_project(monkeypatch, capsys):
+    calls = []
+
+    monkeypatch.setattr(run_crew_module, "_has_json_crew", lambda: False)
+    monkeypatch.setattr(
+        run_crew_module,
+        "read_toml",
+        lambda: {"tool": {"crewai": {"type": "flow"}}},
+    )
+    monkeypatch.setattr(
+        run_crew_module,
+        "_execute_uv_script",
+        lambda script_name, **kwargs: calls.append((script_name, kwargs)),
+    )
+
+    run_crew_module.run_crew()
+
+    assert capsys.readouterr().out == "Running the Flow\n"
+    assert calls == [("kickoff", {"entity_type": "flow"})]
+
+
+def test_run_crew_rejects_filename_for_flow_project(monkeypatch):
+    monkeypatch.setattr(run_crew_module, "_has_json_crew", lambda: False)
+    monkeypatch.setattr(
+        run_crew_module,
+        "read_toml",
+        lambda: {"tool": {"crewai": {"type": "flow"}}},
+    )
+
+    with pytest.raises(click.UsageError) as exc_info:
+        run_crew_module.run_crew(trained_agents_file="trained.pkl")
+
+    assert "--filename can only be used when running crews" in exc_info.value.message
+
+
+def test_run_crew_runs_configured_declarative_flow_project(monkeypatch, capsys):
+    calls = []
+
+    monkeypatch.setattr(run_crew_module, "_has_json_crew", lambda: False)
+    monkeypatch.setattr(
+        run_crew_module,
+        "read_toml",
+        lambda: {
+            "tool": {
+                "crewai": {
+                    "type": "flow",
+                    "definition": "flow.yaml",
+                }
+            }
+        },
+    )
+    monkeypatch.setattr(
+        "crewai_cli.run_declarative_flow.run_declarative_flow_in_project_env",
+        lambda definition, inputs=None: calls.append((definition, inputs)),
+    )
+    monkeypatch.setattr(
+        run_crew_module,
+        "_execute_uv_script",
+        lambda *_args, **_kwargs: pytest.fail("declarative flows must not run kickoff"),
+    )
+
+    run_crew_module.run_crew()
+
+    assert capsys.readouterr().out == "Running the Flow\n"
+    assert calls == [("flow.yaml", None)]

lib/cli/tests/test_run_declarative_flow.py

@@ -83,7 +83,7 @@ def test_run_declarative_flow_in_project_env_uses_uv(
 
     assert subprocess_calls == [
         (
-            ["uv", "run", "crewai", "flow", "kickoff"],
+            ["uv", "run", "crewai", "run"],
             {
                 "capture_output": False,
                 "text": True,

lib/crewai/tests/cli/test_run_crew.py

@@ -11,11 +11,10 @@ import pytest
 
 from crewai_cli.cli import run
 from crewai_cli.run_crew import (
-    CrewType,
+    _execute_uv_script,
     _load_json_crew_for_tui,
     _missing_input_names,
     _prompt_for_missing_inputs,
-    execute_command,
 )
 
 
@@ -30,6 +29,8 @@ def test_run_passes_filename_to_run_crew(run_crew_mock: mock.Mock, runner: CliRu
 
     run_crew_mock.assert_called_once_with(
         trained_agents_file="my_custom_trained.pkl",
+        definition=None,
+        inputs=None,
     )
     assert result.exit_code == 0
 
@@ -38,7 +39,11 @@ def test_run_passes_filename_to_run_crew(run_crew_mock: mock.Mock, runner: CliRu
 def test_run_without_filename_passes_none(run_crew_mock: mock.Mock, runner: CliRunner) -> None:
     result = runner.invoke(run)
 
-    run_crew_mock.assert_called_once_with(trained_agents_file=None)
+    run_crew_mock.assert_called_once_with(
+        trained_agents_file=None,
+        definition=None,
+        inputs=None,
+    )
     assert result.exit_code == 0
 
 
@@ -50,7 +55,11 @@ def test_run_without_filename_passes_none(run_crew_mock: mock.Mock, runner: CliR
 def test_execute_command_sets_env_var_when_filename_provided(
     _build_env: mock.Mock, subprocess_run: mock.Mock
 ) -> None:
-    execute_command(CrewType.STANDARD, trained_agents_file="my_custom_trained.pkl")
+    _execute_uv_script(
+        "run_crew",
+        entity_type="crew",
+        trained_agents_file="my_custom_trained.pkl",
+    )
 
     _, kwargs = subprocess_run.call_args
     assert kwargs["env"]["CREWAI_TRAINED_AGENTS_FILE"] == "my_custom_trained.pkl"
@@ -65,7 +74,7 @@ def test_execute_command_sets_env_var_when_filename_provided(
 def test_execute_command_omits_env_var_when_filename_absent(
     _build_env: mock.Mock, subprocess_run: mock.Mock
 ) -> None:
-    execute_command(CrewType.STANDARD)
+    _execute_uv_script("run_crew", entity_type="crew")
 
     _, kwargs = subprocess_run.call_args
     assert "CREWAI_TRAINED_AGENTS_FILE" not in kwargs["env"]