diff --git a/lib/cli/src/crewai_cli/cli.py b/lib/cli/src/crewai_cli/cli.py index 30eccdc61..28cec9d97 100644 --- a/lib/cli/src/crewai_cli/cli.py +++ b/lib/cli/src/crewai_cli/cli.py @@ -526,8 +526,8 @@ def run( inputs: str | None, ) -> None: """Run the Crew or Flow.""" - if inputs is not None and definition is None: - raise click.UsageError("--inputs requires --definition") + # --inputs no longer requires --definition: with no override it resolves the + # configured [tool.crewai] flow, same as a bare `crewai run`. if trained_agents_file is not None and definition is not None: raise click.UsageError("--filename can only be used when running crews") diff --git a/lib/cli/src/crewai_cli/run_crew.py b/lib/cli/src/crewai_cli/run_crew.py index 8fde69ee8..82173b5d3 100644 --- a/lib/cli/src/crewai_cli/run_crew.py +++ b/lib/cli/src/crewai_cli/run_crew.py @@ -571,9 +571,7 @@ def run_crew( definition: Optional path to a declarative Flow definition. inputs: Optional JSON object passed to a declarative Flow. """ - if inputs is not None and definition is None: - raise click.UsageError("--inputs requires --definition") - + # --definition is a pure override: run that flow directly. if definition is not None: _run_explicit_declarative_flow( definition=definition, @@ -584,6 +582,7 @@ def run_crew( pyproject_data = read_toml() if json_crew_definition := configured_project_json_crew(pyproject_data): + _reject_inputs_for_non_flow(inputs) _run_json_crew_in_project_env( trained_agents_file=trained_agents_file, crew_path=json_crew_definition, @@ -594,18 +593,27 @@ def run_crew( project_type = get_crewai_project_type(pyproject_data) if project_type == "flow": + # No --definition: resolve the configured [tool.crewai] flow — the same + # resolution as a bare `crewai run` — and pass --inputs straight through. _run_flow_project( pyproject_data=pyproject_data, trained_agents_file=trained_agents_file, + inputs=inputs, ) return + _reject_inputs_for_non_flow(inputs) _run_classic_crew_project( pyproject_data=pyproject_data, trained_agents_file=trained_agents_file, ) +def _reject_inputs_for_non_flow(inputs: str | None) -> None: + if inputs is not None: + raise click.UsageError("--inputs is only supported for declarative flows") + + def _run_explicit_declarative_flow( definition: str, inputs: str | None, trained_agents_file: str | None ) -> None: @@ -618,7 +626,9 @@ def _run_explicit_declarative_flow( def _run_flow_project( - pyproject_data: dict[str, Any], trained_agents_file: str | None + pyproject_data: dict[str, Any], + trained_agents_file: str | None, + inputs: str | None = None, ) -> None: if trained_agents_file is not None: raise click.UsageError("--filename can only be used when running crews") @@ -629,9 +639,16 @@ def _run_flow_project( ) if definition := configured_project_declarative_flow(pyproject_data): - run_declarative_flow_in_project_env(definition=definition) + run_declarative_flow_in_project_env(definition=definition, inputs=inputs) return + # No configured declarative flow definition to resolve inputs against. + if inputs is not None: + raise click.UsageError( + "--inputs requires a declarative flow definition " + "([tool.crewai].definition) or --definition" + ) + from crewai_cli.kickoff_flow import ( _load_conversational_flow_from_kickoff_script, _run_conversational_flow_tui, diff --git a/lib/cli/src/crewai_cli/run_declarative_flow.py b/lib/cli/src/crewai_cli/run_declarative_flow.py index 49af0f079..e3f3abad1 100644 --- a/lib/cli/src/crewai_cli/run_declarative_flow.py +++ b/lib/cli/src/crewai_cli/run_declarative_flow.py @@ -1,15 +1,21 @@ from __future__ import annotations +import difflib import json from pathlib import Path import subprocess +import sys from typing import Any import click from crewai_core.project import ProjectDefinitionError, configured_project_definition from pydantic import ValidationError -from crewai_cli.utils import build_env_with_all_tool_credentials +from crewai_cli.utils import ( + build_env_with_all_tool_credentials, + enable_prompt_line_editing, + is_dmn_mode_enabled, +) def run_declarative_flow_in_project_env( @@ -20,10 +26,13 @@ def run_declarative_flow_in_project_env( run_declarative_flow(definition=definition, inputs=inputs) return + # Re-run inside the project env (so the flow loads with the project's deps). + # The configured definition is re-resolved there; forward --inputs so the + # in-env run kicks off with the same values instead of losing them. + command = ["uv", "run", "crewai", "run"] if inputs is not None: - raise click.UsageError("--inputs is only supported with --definition") - - _execute_declarative_flow_command(["uv", "run", "crewai", "run"]) + command += ["--inputs", inputs] + _execute_declarative_flow_command(command) def plot_declarative_flow_in_project_env(definition: str | Path) -> None: @@ -36,12 +45,20 @@ def plot_declarative_flow_in_project_env(definition: str | Path) -> None: def run_declarative_flow(definition: str | Path, inputs: str | None = None) -> None: - """Run a declarative flow from a definition path.""" - parsed_inputs = _parse_inputs(inputs) + """Run a declarative flow from a definition path. + + Inputs come from one place: the flow's own state schema. Any ``--inputs`` + JSON is layered on top as an override, missing required fields are prompted + for interactively, and everything is validated against the schema before + kickoff — so a bare ``crewai run`` on a configured flow just works. + """ + provided = _parse_inputs(inputs) or {} + + flow = load_declarative_flow(definition) + resolved_inputs = _resolve_flow_inputs(flow, provided) try: - flow = load_declarative_flow(definition) - result = flow.kickoff(inputs=parsed_inputs) + result = flow.kickoff(inputs=resolved_inputs or None) except Exception as exc: click.echo( f"An error occurred while running the declarative flow: {exc}", err=True @@ -51,6 +68,158 @@ def run_declarative_flow(definition: str | Path, inputs: str | None = None) -> N click.echo(_format_result(result)) +def _resolve_flow_inputs(flow: Any, provided: dict[str, Any]) -> dict[str, Any]: + """Resolve kickoff inputs from the flow's state schema. + + Warns on unknown keys, prompts for missing required fields (unless + non-interactive), and validates types before kickoff. Exits with a pointed + message when a required input is still missing or an input is invalid. + """ + schema = _flow_state_schema(flow) + if schema is None: + # dict / unschematized state — nothing to derive; pass inputs through. + return dict(provided) + + properties = { + name: spec + for name, spec in (schema.get("properties") or {}).items() + if name != "id" + } + state_model = type(flow.state) + defaults = _flow_state_defaults(flow) + + # Unknown keys are almost always typos — warn and drop them (they'd fail + # structured-state validation at kickoff anyway). + collected: dict[str, Any] = {} + for key, value in provided.items(): + if key in properties: + collected[key] = value + continue + suggestion = _closest_key(key, properties) + hint = f" Did you mean '{suggestion}'?" if suggestion else "" + click.secho( + f" Ignoring unknown input '{key}' — not in the flow's state schema.{hint}", + fg="yellow", + err=True, + ) + + missing = _missing_required(state_model, {**defaults, **collected}) + if missing and _is_interactive(): + collected.update(_prompt_for_flow_inputs(missing, properties)) + missing = _missing_required(state_model, {**defaults, **collected}) + + if missing: + for name in missing: + description = (properties.get(name) or {}).get("description") + suffix = f" — {description}" if description else "" + click.secho(f" Missing required input '{name}'{suffix}", fg="red", err=True) + raise SystemExit(1) + + _validate_flow_inputs(state_model, {**defaults, **collected}) + return collected + + +def _is_interactive() -> bool: + """Prompt only in an interactive terminal, never in non-interactive mode.""" + return not is_dmn_mode_enabled() and sys.stdin.isatty() + + +def _flow_state_schema(flow: Any) -> dict[str, Any] | None: + """Return the flow's state JSON schema, or ``None`` for dict/plain state.""" + state = getattr(flow, "state", None) + if state is None or isinstance(state, dict): + return None + model_json_schema = getattr(type(state), "model_json_schema", None) + if not callable(model_json_schema): + return None + try: + schema = model_json_schema() + except Exception: + return None + return schema if isinstance(schema, dict) else None + + +def _flow_state_defaults(flow: Any) -> dict[str, Any]: + """Declared state defaults (``state.default``) from the flow definition.""" + state_definition = getattr(getattr(flow, "_definition", None), "state", None) + default = getattr(state_definition, "default", None) + return dict(default) if isinstance(default, dict) else {} + + +def _missing_required(state_model: Any, values: dict[str, Any]) -> list[str]: + """Required state fields not satisfied by ``values`` (defaults + inputs).""" + try: + state_model.model_validate(values) + except ValidationError as exc: + return [ + str(error["loc"][0]) + for error in exc.errors() + if error.get("type") == "missing" and error.get("loc") + ] + return [] + + +def _validate_flow_inputs(state_model: Any, values: dict[str, Any]) -> None: + """Validate inputs against the state schema; exit with pointed type errors.""" + try: + state_model.model_validate(values) + except ValidationError as exc: + for error in exc.errors(): + location = ".".join(str(part) for part in error.get("loc", ())) + click.secho( + f" Invalid input '{location}': {error.get('msg')}", fg="red", err=True + ) + raise SystemExit(1) from exc + + +def _prompt_for_flow_inputs( + missing: list[str], properties: dict[str, Any] +) -> dict[str, Any]: + """Prompt for each missing required field, showing its schema description.""" + enable_prompt_line_editing() + # Prompt chrome goes to stderr so stdout carries only the flow result. + click.echo(err=True) + click.secho(" Flow inputs", fg="cyan", bold=True, err=True) + click.secho(" This flow needs the following to run.", dim=True, err=True) + + collected: dict[str, Any] = {} + for name in missing: + spec = properties.get(name) or {} + description = spec.get("description") + if description: + click.secho(f" {description}", dim=True, err=True) + raw = click.prompt( + click.style(f" {name}", fg="cyan"), + prompt_suffix=click.style(" > ", fg="bright_white"), + ) + collected[name] = _coerce_input(raw, spec) + return collected + + +def _coerce_input(raw: str, spec: dict[str, Any]) -> Any: + """Best-effort coerce a prompted string to the field's JSON-schema type.""" + field_type = spec.get("type") + if field_type == "integer": + try: + return int(raw) + except ValueError: + return raw + if field_type == "number": + try: + return float(raw) + except ValueError: + return raw + if field_type == "boolean": + return raw.strip().lower() in {"1", "true", "yes", "y", "on"} + return raw + + +def _closest_key(key: str, properties: dict[str, Any]) -> str | None: + """Nearest schema field name to a likely typo, if one is close enough.""" + matches = difflib.get_close_matches(key, list(properties), n=1, cutoff=0.7) + return matches[0] if matches else None + + def plot_declarative_flow(definition: str | Path) -> None: """Plot a declarative flow from a definition path.""" try: diff --git a/lib/cli/tests/test_cli.py b/lib/cli/tests/test_cli.py index acdeee7ff..535daf7f2 100644 --- a/lib/cli/tests/test_cli.py +++ b/lib/cli/tests/test_cli.py @@ -151,12 +151,14 @@ def test_run_with_definition_uses_project_runner(run_crew, runner): @mock.patch("crewai_cli.cli.run_crew") -def test_run_rejects_inputs_without_definition(run_crew, runner): +def test_run_inputs_without_definition_calls_run_crew(run_crew, runner): + # --inputs no longer requires --definition; the resolution happens in run_crew. result = runner.invoke(run, ["--inputs", '{"topic":"AI"}']) - assert result.exit_code == 2 - assert "Error: --inputs requires --definition" in result.output - run_crew.assert_not_called() + assert result.exit_code == 0 + run_crew.assert_called_once_with( + trained_agents_file=None, definition=None, inputs='{"topic":"AI"}' + ) @mock.patch("crewai_cli.cli.run_crew") diff --git a/lib/cli/tests/test_run_crew.py b/lib/cli/tests/test_run_crew.py index a1b4c73bc..3c7a4ac2e 100644 --- a/lib/cli/tests/test_run_crew.py +++ b/lib/cli/tests/test_run_crew.py @@ -682,11 +682,48 @@ def test_configured_project_json_crew_ignores_missing_pyproject( assert run_crew_module.configured_project_json_crew() is None -def test_run_crew_rejects_inputs_without_definition(): +def test_run_crew_inputs_without_definition_rejected_for_non_flow(monkeypatch): + # --inputs is flow-only; in a non-flow project it now errors clearly instead + # of the old "--inputs requires --definition". + monkeypatch.setattr(run_crew_module, "read_toml", lambda *a, **k: {}) + monkeypatch.setattr( + run_crew_module, "configured_project_json_crew", lambda *a, **k: None + ) + monkeypatch.setattr( + run_crew_module, "_warn_if_old_poetry_project", lambda *a, **k: None + ) + monkeypatch.setattr(run_crew_module, "get_crewai_project_type", lambda *a, **k: "crew") + 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 + assert "--inputs is only supported for declarative flows" in exc_info.value.message + + +def test_run_crew_inputs_without_definition_resolves_configured_flow(monkeypatch): + # --inputs with no --definition resolves the configured [tool.crewai] flow, + # exactly like a bare `crewai run`, and forwards the inputs. + import crewai_cli.run_declarative_flow as rdf + + calls: dict[str, object] = {} + monkeypatch.setattr(run_crew_module, "read_toml", lambda *a, **k: {}) + monkeypatch.setattr( + run_crew_module, "configured_project_json_crew", lambda *a, **k: None + ) + monkeypatch.setattr( + run_crew_module, "_warn_if_old_poetry_project", lambda *a, **k: None + ) + monkeypatch.setattr(run_crew_module, "get_crewai_project_type", lambda *a, **k: "flow") + monkeypatch.setattr( + rdf, "configured_project_declarative_flow", lambda *a, **k: Path("flow.yaml") + ) + monkeypatch.setattr( + rdf, "run_declarative_flow_in_project_env", lambda **kw: calls.update(kw) + ) + + run_crew_module.run_crew(inputs='{"topic":"AI"}') + + assert calls == {"definition": Path("flow.yaml"), "inputs": '{"topic":"AI"}'} def test_run_crew_rejects_filename_with_explicit_definition(): diff --git a/lib/cli/tests/test_run_declarative_flow.py b/lib/cli/tests/test_run_declarative_flow.py index 8d5435c8f..396578cd6 100644 --- a/lib/cli/tests/test_run_declarative_flow.py +++ b/lib/cli/tests/test_run_declarative_flow.py @@ -146,3 +146,192 @@ def test_run_declarative_flow_in_process_inside_uv( ) assert capsys.readouterr().out == "AI\n" + + +def test_run_declarative_flow_in_project_env_forwards_inputs( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + subprocess_calls = [] + monkeypatch.chdir(tmp_path) + monkeypatch.delenv("UV_RUN_RECURSION_DEPTH", raising=False) + (tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n") + monkeypatch.setattr( + run_declarative_flow_module, + "build_env_with_all_tool_credentials", + lambda: {}, + ) + monkeypatch.setattr( + run_declarative_flow_module.subprocess, + "run", + lambda command, **kwargs: subprocess_calls.append(command), + ) + + run_declarative_flow_module.run_declarative_flow_in_project_env( + "flow.yaml", '{"topic":"AI"}' + ) + + # --inputs is forwarded to the in-env run instead of being rejected. + assert subprocess_calls == [ + ["uv", "run", "crewai", "run", "--inputs", '{"topic":"AI"}'] + ] + + +# ── Schema-driven inputs: prompt, validate, override ──────────────── + +REQUIRED_FLOW_YAML = """\ +schema: crewai.flow/v1 +name: RequiredInputFlow +config: + suppress_flow_events: true +state: + type: json_schema + json_schema: + type: object + properties: + prospect_email: + type: string + description: Email address of the prospect to research + required: [prospect_email] +methods: + begin: + start: true + do: + call: expression + expr: state.prospect_email +""" + +DEFAULTS_FLOW_YAML = """\ +schema: crewai.flow/v1 +name: DefaultsFlow +config: + suppress_flow_events: true +state: + type: json_schema + json_schema: + type: object + properties: + topic: {type: string} + audience: {type: string} + required: [topic, audience] + default: + topic: AI +methods: + begin: + start: true + do: + call: expression + expr: state.audience +""" + +TYPED_FLOW_YAML = """\ +schema: crewai.flow/v1 +name: TypedFlow +config: + suppress_flow_events: true +state: + type: json_schema + json_schema: + type: object + properties: + count: {type: integer} + required: [count] +methods: + begin: + start: true + do: + call: expression + expr: state.count +""" + + +def _write(tmp_path: Path, contents: str) -> Path: + path = tmp_path / "flow.yaml" + path.write_text(contents, encoding="utf-8") + return path + + +def test_inputs_flag_satisfies_required_field( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + path = _write(tmp_path, REQUIRED_FLOW_YAML) + + run_declarative_flow_module.run_declarative_flow( + str(path), '{"prospect_email":"a@b.com"}' + ) + + assert capsys.readouterr().out == "a@b.com\n" + + +def test_missing_required_reports_pointed_error( + tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(run_declarative_flow_module, "_is_interactive", lambda: False) + path = _write(tmp_path, REQUIRED_FLOW_YAML) + + with pytest.raises(SystemExit): + run_declarative_flow_module.run_declarative_flow(str(path)) + + assert ( + "Missing required input 'prospect_email' — " + "Email address of the prospect to research" in capsys.readouterr().err + ) + + +def test_prompts_for_missing_required_when_interactive( + tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + path = _write(tmp_path, REQUIRED_FLOW_YAML) + monkeypatch.setattr(run_declarative_flow_module, "_is_interactive", lambda: True) + prompted: list[str] = [] + + def fake_prompt(text: str, **kwargs: object) -> str: + prompted.append(text) + return "typed@example.com" + + monkeypatch.setattr(run_declarative_flow_module.click, "prompt", fake_prompt) + + run_declarative_flow_module.run_declarative_flow(str(path)) + + assert capsys.readouterr().out == "typed@example.com\n" + assert any("prospect_email" in text for text in prompted) + + +def test_defaults_satisfy_required_and_are_not_prompted( + tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(run_declarative_flow_module, "_is_interactive", lambda: False) + path = _write(tmp_path, DEFAULTS_FLOW_YAML) + + with pytest.raises(SystemExit): + run_declarative_flow_module.run_declarative_flow(str(path)) + + err = capsys.readouterr().err + # topic has a state default -> satisfied; only audience is missing. + assert "Missing required input 'audience'" in err + assert "'topic'" not in err + + +def test_warns_on_unknown_input_with_suggestion( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + path = _write(tmp_path, REQUIRED_FLOW_YAML) + + run_declarative_flow_module.run_declarative_flow( + str(path), '{"prospect_email":"a@b.com","prospect_emai":"typo"}' + ) + + captured = capsys.readouterr() + assert captured.out == "a@b.com\n" + assert "Ignoring unknown input 'prospect_emai'" in captured.err + assert "Did you mean 'prospect_email'?" in captured.err + + +def test_validates_input_types_before_kickoff( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + path = _write(tmp_path, TYPED_FLOW_YAML) + + with pytest.raises(SystemExit): + run_declarative_flow_module.run_declarative_flow(str(path), '{"count":"nope"}') + + assert "Invalid input 'count'" in capsys.readouterr().err