fix(cli): unify crewai run flow input resolution; prompt from state schema

`crewai run` resolved the configured [tool.crewai] flow, but `--inputs` was
hard-gated behind `--definition` and routed through a separate branch — the two
ways of pointing at the same flow didn't share resolution, and required inputs
were never detected, prompted, or validated (a missing field only blew up at
runtime).

Now inputs and definition come from one place:

- Remove the "--inputs requires --definition" gate (cli.py, run_crew.py,
  run_declarative_flow.py). `--inputs` alone resolves the configured flow,
  exactly like a bare `crewai run`; `--definition` is purely an override. The
  project-env re-exec forwards `--inputs` instead of rejecting it.
- Read the flow's state schema from the runtime Flow instance
  (`type(flow.state).model_json_schema()`), which is reliable for both inline
  `json_schema` and ref-imported `pydantic` state (the static definition's
  json_schema is None for the common ref case).
- Plain `crewai run` detects required state fields (minus those satisfied by
  state defaults) and prompts for them interactively, showing each field's
  description; skipped in non-interactive / CREWAI_DMN mode.
- Validate against the schema before kickoff: pointed
  "Missing required input 'x' — <description>" errors, and warn on unknown keys
  with a did-you-mean suggestion (catches typos like `prospect_emai`).

`--inputs` on a non-flow project now errors clearly ("only supported for
declarative flows") instead of the old confusing gate.

Tests: schema-driven prompt/validate/override paths, unknown-key warning,
defaults-satisfy-required, type validation, and re-exec input forwarding.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RBYGqJHC2TMC6fonFziuuh

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")
 

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,

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:

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")

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():

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