andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-07-08 16:35:08 +00:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

feat(cli): unify runtime-input prompting across declarative flows and crews

Browse Source 
Declarative (JSON) crews now resolve inputs the same way declarative flows
do, via a shared crewai_cli.input_prompt module (prompt_for_inputs,
parse_inputs_json, closest_name, is_interactive):

- accept --inputs (previously rejected for crews), forwarded to the crew
  subprocess via CREWAI_JSON_CREW_INPUTS and validated before spinning up uv
- layer --inputs over the crew's declared `inputs` defaults
- prompt for missing {placeholder}s with the same UX as flows, and error
  cleanly with a pointed per-name message when non-interactive
- warn on unknown keys with a "did you mean" suggestion

Unlike flows — whose state schema is authoritative, so unknown keys are
dropped — the crew placeholder scan is heuristic (agent/task text fields
only), so unrecognized keys are warned about but kept, to avoid discarding a
value a field the scan doesn't cover may rely on.

--inputs remains rejected for classic (Python/YAML) crews, which take their
inputs from main.py. run_declarative_flow's private input helpers move to the
shared module with no behavior change.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RBYGqJHC2TMC6fonFziuuh
This commit is contained in:
Joao Moura
2026-07-07 12:03:54 -07:00
parent 792ca8ec75
commit 96546e37c3
6 changed files with 470 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

84
lib/cli/src/crewai_cli/input_prompt.py Normal file
View File

@@ -0,0 +1,84 @@
"""Shared interactive prompting for runtime inputs (flows and crews).
``crewai run`` asks the user for values that were not provided up front — for a
declarative flow (derived from its state schema) and for a declarative (JSON)
crew (derived from the ``{placeholder}`` references in its agents and tasks).
Both paths go through this module so the experience is identical: the same
header, the same per-field prompt styling, and prompt chrome on stderr so
stdout carries only the run's result.
"""
from __future__ import annotations
from collections.abc import Callable, Iterable
import difflib
import json
import sys
from typing import Any
import click
from crewai_cli.utils import enable_prompt_line_editing, is_dmn_mode_enabled
def parse_inputs_json(inputs: str | None) -> dict[str, Any] | None:
"""Parse a ``--inputs`` JSON object, exiting with a pointed error if invalid."""
if inputs is None:
return None
try:
parsed = json.loads(inputs)
except json.JSONDecodeError as exc:
click.echo(f"Invalid --inputs JSON: {exc}", err=True)
raise SystemExit(1) from exc
if not isinstance(parsed, dict):
click.echo("Invalid --inputs JSON: expected an object.", err=True)
raise SystemExit(1)
return parsed
def closest_name(key: str, candidates: Iterable[str]) -> str | None:
"""Nearest candidate name to a likely typo, if one is close enough."""
matches = difflib.get_close_matches(key, list(candidates), n=1, cutoff=0.7)
return matches[0] if matches else None
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 prompt_for_inputs(
names: list[str],
*,
title: str,
subtitle: str,
describe: Callable[[str], str | None] | None = None,
coerce: Callable[[str, str], Any] | None = None,
) -> dict[str, Any]:
"""Prompt for each name and return ``{name: value}``.
``describe(name)`` returns an optional hint shown dim above the prompt (used
by flows to surface a field's schema description). ``coerce(name, raw)``
converts the typed string to the stored value (used by flows to coerce to
the field's JSON-schema type); by default the raw string is kept as-is.
Prompt chrome is written to stderr so stdout carries only the run result.
"""
enable_prompt_line_editing()
click.echo(err=True)
click.secho(f" {title}", fg="cyan", bold=True, err=True)
click.secho(f" {subtitle}", dim=True, err=True)
collected: dict[str, Any] = {}
for name in names:
if describe is not None and (hint := describe(name)):
click.secho(f" {hint}", dim=True, err=True)
raw = click.prompt(
click.style(f" {name}", fg="cyan"),
prompt_suffix=click.style(" > ", fg="bright_white"),
)
collected[name] = coerce(name, raw) if coerce is not None else raw
return collected

145
lib/cli/src/crewai_cli/run_crew.py
View File

@@ -12,9 +12,14 @@ import click
from crewai_core.constants import CREWAI_TRAINED_AGENTS_FILE_ENV
from packaging import version
from crewai_cli.input_prompt import (
closest_name,
is_interactive,
parse_inputs_json,
prompt_for_inputs,
)
from crewai_cli.utils import (
build_env_with_all_tool_credentials,
enable_prompt_line_editing,
is_dmn_mode_enabled,
)
from crewai_cli.version import get_crewai_tools_dependency, get_crewai_version
@@ -31,6 +36,7 @@ _INPUT_PLACEHOLDER_RE = re.compile(r"(?<!{){([A-Za-z_][A-Za-z0-9_\-]*)}(?!})")
_CREWAI_CLI_RUNNER_PACKAGE_DIR_ENV = "CREWAI_CLI_RUNNER_PACKAGE_DIR"
_CREWAI_RUNNER_SOURCE_DIR_ENV = "CREWAI_RUNNER_SOURCE_DIR"
_CREWAI_JSON_CREW_DEFINITION_ENV = "CREWAI_JSON_CREW_DEFINITION"
_CREWAI_JSON_CREW_INPUTS_ENV = "CREWAI_JSON_CREW_INPUTS"
_FULL_CREWAI_INSTALL_MESSAGE = f"""\
CrewAI CLI is installed without the `crewai` package required to run crews.
@@ -79,6 +85,8 @@ kwargs = {
}
if crew_definition := os.getenv("CREWAI_JSON_CREW_DEFINITION"):
kwargs["crew_path"] = crew_definition
if crew_inputs := os.getenv("CREWAI_JSON_CREW_INPUTS"):
kwargs["inputs"] = crew_inputs
try:
module._run_json_crew(**kwargs)
@@ -138,8 +146,8 @@ def _extract_input_placeholders(text: str | None) -> set[str]:
return set(_INPUT_PLACEHOLDER_RE.findall(text))
def _missing_input_names(crew: Any, inputs: dict[str, Any]) -> list[str]:
"""Return input placeholders used by a crew but not provided as defaults."""
def _referenced_input_names(crew: Any) -> set[str]:
"""All ``{placeholder}`` names referenced by a crew's agents and tasks."""
placeholders: set[str] = set()
for agent in getattr(crew, "agents", []) or []:
@@ -160,32 +168,70 @@ def _missing_input_names(crew: Any, inputs: dict[str, Any]) -> list[str]:
_extract_input_placeholders(getattr(task, "output_file", None))
)
return sorted(name for name in placeholders if name not in inputs)
return placeholders
def _prompt_for_missing_inputs(
crew: Any, default_inputs: dict[str, Any]
def _missing_input_names(crew: Any, inputs: dict[str, Any]) -> list[str]:
"""Return input placeholders referenced by a crew but not provided as inputs."""
return sorted(name for name in _referenced_input_names(crew) if name not in inputs)
def _resolve_crew_inputs(
crew: Any,
default_inputs: dict[str, Any],
provided: dict[str, Any] | None,
*,
interactive: bool,
) -> dict[str, Any]:
"""Ask for runtime values for placeholders that lack default inputs."""
"""Resolve kickoff inputs for a declarative crew.
Mirrors the declarative-flow experience (``_resolve_flow_inputs``): layers
``--inputs`` over the crew's declared ``inputs`` defaults, warns on provided
keys that aren't referenced as ``{placeholder}``s, prompts for any
still-missing placeholders when interactive, and exits with a pointed
message when one is still missing.
Unlike flows — whose state schema is an authoritative contract, so unknown
keys are dropped — the crew placeholder scan is heuristic (it only covers
agent/task text fields). An unrecognized key is therefore warned about but
*kept*, never dropped: dropping could silently discard a value that a field
the scan doesn't cover actually relies on.
"""
referenced = _referenced_input_names(crew)
inputs = dict(default_inputs or {})
for key, value in (provided or {}).items():
if key not in referenced:
suggestion = closest_name(key, referenced)
hint = f" Did you mean '{suggestion}'?" if suggestion else ""
click.secho(
f" Input '{key}' isn't referenced by any {{placeholder}} "
f"in the crew.{hint}",
fg="yellow",
err=True,
)
inputs[key] = value
missing = _missing_input_names(crew, inputs)
if not missing:
return inputs
enable_prompt_line_editing()
click.echo()
click.secho(" Runtime inputs", fg="cyan", bold=True)
click.secho(
" Values for {placeholder} references in your agents and tasks.",
dim=True,
)
for name in missing:
inputs[name] = click.prompt(
click.style(f" {name}", fg="cyan"),
prompt_suffix=click.style(" > ", fg="bright_white"),
if missing and interactive:
inputs.update(
prompt_for_inputs(
missing,
title="Crew inputs",
subtitle="This crew needs the following to run.",
)
)
missing = _missing_input_names(crew, inputs)
if missing:
for name in missing:
click.secho(f" Missing required input '{name}'", fg="red", err=True)
click.secho(
" Provide them via --inputs or the `inputs` object in crew.json(c).",
dim=True,
err=True,
)
raise SystemExit(1)
return inputs
@@ -243,29 +289,14 @@ def _prepare_json_crew_for_tui(crew: Any) -> None:
agent.llm.stream = True
def _runtime_inputs_without_prompt(
crew: Any, default_inputs: dict[str, Any]
) -> dict[str, Any]:
"""Return runtime inputs in non-interactive mode or exit on missing values."""
inputs = dict(default_inputs or {})
missing = _missing_input_names(crew, inputs)
if missing:
missing_list = ", ".join(missing)
click.echo(
"Missing runtime inputs for CREWAI_DMN mode: "
f"{missing_list}. Add them to the `inputs` object in crew.json(c).",
err=True,
)
raise SystemExit(1)
return inputs
def _run_json_crew_without_tui(crew_path: Path) -> Any:
def _run_json_crew_without_tui(crew_path: Path, provided: dict[str, Any] | None) -> Any:
"""Run a JSON-defined crew with plain terminal output."""
with _json_loading_status("Preparing crew..."):
crew, default_inputs = _load_json_crew(crew_path)
runtime_inputs = _runtime_inputs_without_prompt(crew, default_inputs)
runtime_inputs = _resolve_crew_inputs(
crew, default_inputs, provided, interactive=False
)
result = crew.kickoff(inputs=runtime_inputs)
if result is not None:
click.echo(str(result))
@@ -275,6 +306,7 @@ def _run_json_crew_without_tui(crew_path: Path) -> Any:
def _run_json_crew(
trained_agents_file: str | None = None,
crew_path: str | Path | None = None,
inputs: str | None = None,
) -> Any:
"""Load and run a JSON-defined crew."""
from dotenv import load_dotenv
@@ -296,13 +328,17 @@ def _run_json_crew(
)
crew_path = Path(crew_path)
provided = parse_inputs_json(inputs)
if is_dmn_mode_enabled():
return _run_json_crew_without_tui(crew_path)
return _run_json_crew_without_tui(crew_path, provided)
crew_run_app_cls, crew, default_inputs, task_names, agent_names = (
_load_json_crew_for_tui(crew_path)
)
runtime_inputs = _prompt_for_missing_inputs(crew, default_inputs)
runtime_inputs = _resolve_crew_inputs(
crew, default_inputs, provided, interactive=is_interactive()
)
app = crew_run_app_cls(
crew_name=crew.name or "Crew",
@@ -411,12 +447,18 @@ def _json_crew_run_command(project_root: Path | None = None) -> list[str]:
def _run_json_crew_in_project_env(
trained_agents_file: str | None = None,
crew_path: str | Path | None = None,
inputs: str | None = None,
) -> Any:
"""Run JSON crews from the project's uv-managed environment."""
# Validate --inputs up front so bad JSON fails before we spin up the uv env.
if inputs is not None:
parse_inputs_json(inputs)
if not (Path.cwd() / "pyproject.toml").is_file():
return _run_json_crew(
trained_agents_file=trained_agents_file,
crew_path=crew_path,
inputs=inputs,
)
_install_json_crew_dependencies_if_needed()
@@ -430,6 +472,8 @@ def _run_json_crew_in_project_env(
env[CREWAI_TRAINED_AGENTS_FILE_ENV] = trained_agents_file
if crew_path is not None:
env[_CREWAI_JSON_CREW_DEFINITION_ENV] = str(crew_path)
if inputs is not None:
env[_CREWAI_JSON_CREW_INPUTS_ENV] = inputs
try:
subprocess.run( # noqa: S603
@@ -569,7 +613,9 @@ def run_crew(
``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.
inputs: Optional JSON object of runtime inputs for a declarative flow
or declarative (JSON) crew. Layered over the definition's own
defaults; missing required values are prompted for interactively.
"""
# --definition is a pure override: run that flow directly.
if definition is not None:
@@ -582,10 +628,13 @@ def run_crew(
pyproject_data = read_toml()
if json_crew_definition := configured_project_json_crew(pyproject_data):
_reject_inputs_for_non_flow(inputs)
# Declarative (JSON) crews resolve inputs the same way flows do: --inputs
# layers over the crew's declared defaults, missing {placeholder}s are
# prompted for, and unknown keys are flagged. Forward the raw JSON.
_run_json_crew_in_project_env(
trained_agents_file=trained_agents_file,
crew_path=json_crew_definition,
inputs=inputs,
)
return
@@ -611,7 +660,9 @@ def run_crew(
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")
raise click.UsageError(
"--inputs is only supported for declarative flows and crews"
)
def _run_explicit_declarative_flow(

75
lib/cli/src/crewai_cli/run_declarative_flow.py
View File

@@ -1,21 +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,
enable_prompt_line_editing,
is_dmn_mode_enabled,
from crewai_cli.input_prompt import (
closest_name,
is_interactive,
parse_inputs_json,
prompt_for_inputs,
)
from crewai_cli.utils import build_env_with_all_tool_credentials
def run_declarative_flow_in_project_env(
@@ -61,7 +61,7 @@ def run_declarative_flow(definition: str | Path, inputs: str | None = None) -> N
if env_file.exists():
load_dotenv(env_file, override=True)
provided = _parse_inputs(inputs) or {}
provided = parse_inputs_json(inputs) or {}
flow = load_declarative_flow(definition)
resolved_inputs = _resolve_flow_inputs(flow, provided)
@@ -111,7 +111,7 @@ def _resolve_flow_inputs(flow: Any, provided: dict[str, Any]) -> dict[str, Any]:
if key in properties:
collected[key] = value
continue
suggestion = _closest_key(key, properties)
suggestion = closest_name(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}",
@@ -121,7 +121,15 @@ def _resolve_flow_inputs(flow: Any, provided: dict[str, Any]) -> dict[str, Any]:
missing = _missing_required(state_model, {**defaults, **collected})
if missing and _is_interactive():
collected.update(_prompt_for_flow_inputs(missing, properties))
collected.update(
prompt_for_inputs(
missing,
title="Flow inputs",
subtitle="This flow needs the following to run.",
describe=lambda name: (properties.get(name) or {}).get("description"),
coerce=lambda name, raw: _coerce_input(raw, properties.get(name) or {}),
)
)
missing = _missing_required(state_model, {**defaults, **collected})
if missing:
@@ -139,7 +147,7 @@ def _resolve_flow_inputs(flow: Any, provided: dict[str, Any]) -> dict[str, Any]:
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()
return is_interactive()
def _flow_state_schema(flow: Any) -> dict[str, Any] | None:
@@ -190,30 +198,6 @@ def _validate_flow_inputs(state_model: Any, values: dict[str, Any]) -> None:
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")
@@ -232,12 +216,6 @@ def _coerce_input(raw: str, spec: dict[str, Any]) -> Any:
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:
@@ -339,23 +317,6 @@ def _has_project_file(project_root: Path | None = None) -> bool:
return (root / "pyproject.toml").is_file()
def _parse_inputs(inputs: str | None) -> dict[str, Any] | None:
if inputs is None:
return None
try:
parsed = json.loads(inputs)
except json.JSONDecodeError as exc:
click.echo(f"Invalid --inputs JSON: {exc}", err=True)
raise SystemExit(1) from exc
if not isinstance(parsed, dict):
click.echo("Invalid --inputs JSON: expected an object.", err=True)
raise SystemExit(1)
return parsed
def _format_result(result: Any) -> str:
raw_result = getattr(result, "raw", result)
if isinstance(raw_result, str):

83
lib/cli/tests/test_input_prompt.py Normal file
View File

@@ -0,0 +1,83 @@
"""Tests for the shared runtime-input prompting used by flows and crews."""
from __future__ import annotations
import pytest
import crewai_cli.input_prompt as input_prompt_module
from crewai_cli.input_prompt import (
closest_name,
parse_inputs_json,
prompt_for_inputs,
)
def test_parse_inputs_json_returns_none_for_none():
assert parse_inputs_json(None) is None
def test_parse_inputs_json_parses_object():
assert parse_inputs_json('{"topic": "AI"}') == {"topic": "AI"}
def test_parse_inputs_json_rejects_invalid_json(capsys):
with pytest.raises(SystemExit) as exc_info:
parse_inputs_json("not json")
assert exc_info.value.code == 1
assert "Invalid --inputs JSON" in capsys.readouterr().err
def test_parse_inputs_json_rejects_non_object(capsys):
with pytest.raises(SystemExit) as exc_info:
parse_inputs_json("[1, 2, 3]")
assert exc_info.value.code == 1
assert "expected an object" in capsys.readouterr().err
def test_closest_name_suggests_near_miss():
assert closest_name("prospect_emai", ["prospect_email", "topic"]) == "prospect_email"
def test_closest_name_returns_none_when_nothing_close():
assert closest_name("zzzzz", ["prospect_email", "topic"]) is None
def test_prompt_for_inputs_uses_describe_and_coerce(monkeypatch, capsys):
seen: list[str] = []
def fake_prompt(text: str, **kwargs: object) -> str:
seen.append(text)
return "42"
monkeypatch.setattr(input_prompt_module.click, "prompt", fake_prompt)
result = prompt_for_inputs(
["count"],
title="Flow inputs",
subtitle="This flow needs the following to run.",
describe=lambda name: f"How many {name}?",
coerce=lambda name, raw: int(raw),
)
captured = capsys.readouterr()
assert result == {"count": 42}
assert any("count" in text for text in seen)
# Header, subtitle, and description hint all render on stderr.
assert "Flow inputs" in captured.err
assert "How many count?" in captured.err
def test_prompt_for_inputs_keeps_raw_string_without_coerce(monkeypatch):
monkeypatch.setattr(
input_prompt_module.click, "prompt", lambda text, **kwargs: "AI"
)
result = prompt_for_inputs(
["topic"],
title="Crew inputs",
subtitle="This crew needs the following to run.",
)
assert result == {"topic": "AI"}

194
lib/cli/tests/test_run_crew.py
View File

@@ -9,6 +9,7 @@ import click
import pytest
from crewai_core.constants import CREWAI_TRAINED_AGENTS_FILE_ENV
import crewai_cli.input_prompt as input_prompt_module
import crewai_cli.run_crew as run_crew_module
@@ -44,9 +45,12 @@ def test_run_crew_forwards_trained_agents_file_to_json_crews(monkeypatch):
)
called: dict = {}
def fake_run_json_crew_in_project_env(trained_agents_file=None, crew_path=None):
def fake_run_json_crew_in_project_env(
trained_agents_file=None, crew_path=None, inputs=None
):
called["trained_agents_file"] = trained_agents_file
called["crew_path"] = crew_path
called["inputs"] = inputs
monkeypatch.setattr(
run_crew_module,
@@ -59,6 +63,7 @@ def test_run_crew_forwards_trained_agents_file_to_json_crews(monkeypatch):
assert called == {
"trained_agents_file": "some.pkl",
"crew_path": Path("crew.jsonc"),
"inputs": None,
}
@@ -305,9 +310,10 @@ def test_json_run_without_pyproject_runs_in_process(monkeypatch, tmp_path: Path)
monkeypatch.chdir(tmp_path)
called: dict = {}
def fake_run_json_crew(trained_agents_file=None, crew_path=None):
def fake_run_json_crew(trained_agents_file=None, crew_path=None, inputs=None):
called["trained_agents_file"] = trained_agents_file
called["crew_path"] = crew_path
called["inputs"] = inputs
return "result"
monkeypatch.setattr(run_crew_module, "_run_json_crew", fake_run_json_crew)
@@ -318,7 +324,11 @@ def test_json_run_without_pyproject_runs_in_process(monkeypatch, tmp_path: Path)
)
== "result"
)
assert called == {"trained_agents_file": "trained.pkl", "crew_path": None}
assert called == {
"trained_agents_file": "trained.pkl",
"crew_path": None,
"inputs": None,
}
def test_json_project_env_run_failure_exits_nonzero(monkeypatch, tmp_path: Path):
@@ -535,7 +545,9 @@ def _patch_tui_run(monkeypatch, status: str):
lambda _path: (FakeApp, crew, {}, [], []),
)
monkeypatch.setattr(
run_crew_module, "_prompt_for_missing_inputs", lambda _crew, inputs: inputs
run_crew_module,
"_resolve_crew_inputs",
lambda _crew, default_inputs, _provided, *, interactive: default_inputs,
)
monkeypatch.setattr(run_crew_module, "_print_post_tui_summary", lambda _app: None)
@@ -636,7 +648,168 @@ def test_run_json_crew_dmn_mode_exits_on_missing_inputs(
captured = capsys.readouterr()
assert exc_info.value.code == 1
assert "Missing runtime inputs for CREWAI_DMN mode: topic" in captured.err
assert "Missing required input 'topic'" in captured.err
# ── Declarative-crew inputs: merge --inputs, warn, prompt (flow parity) ──
def _crew_with_placeholders(*names: str) -> object:
"""A minimal crew whose agent goal references each ``{name}`` placeholder."""
from types import SimpleNamespace
goal = " ".join(f"{{{name}}}" for name in names)
return SimpleNamespace(
agents=[SimpleNamespace(role="Researcher", goal=goal, backstory="")],
tasks=[],
)
def test_resolve_crew_inputs_merges_inputs_over_defaults():
crew = _crew_with_placeholders("topic")
resolved = run_crew_module._resolve_crew_inputs(
crew, {"topic": "AI"}, {"topic": "ML"}, interactive=False
)
assert resolved == {"topic": "ML"}
def test_resolve_crew_inputs_warns_and_keeps_unknown_input(capsys):
# The placeholder scan is heuristic, so an unreferenced key is flagged but
# kept (never silently dropped like a flow's schema would).
crew = _crew_with_placeholders("topic")
resolved = run_crew_module._resolve_crew_inputs(
crew, {}, {"topic": "AI", "topi": "typo"}, interactive=False
)
captured = capsys.readouterr()
assert resolved == {"topic": "AI", "topi": "typo"}
assert "isn't referenced by any {placeholder}" in captured.err
assert "Did you mean 'topic'?" in captured.err
def test_resolve_crew_inputs_prompts_when_interactive(monkeypatch):
crew = _crew_with_placeholders("topic")
prompted: list[str] = []
def fake_prompt(text: str, **kwargs: object) -> str:
prompted.append(text)
return "AI"
monkeypatch.setattr(input_prompt_module.click, "prompt", fake_prompt)
resolved = run_crew_module._resolve_crew_inputs(
crew, {}, None, interactive=True
)
assert resolved == {"topic": "AI"}
assert any("topic" in text for text in prompted)
def test_resolve_crew_inputs_errors_when_missing_non_interactive(capsys):
crew = _crew_with_placeholders("topic")
with pytest.raises(SystemExit) as exc_info:
run_crew_module._resolve_crew_inputs(crew, {}, None, interactive=False)
assert exc_info.value.code == 1
assert "Missing required input 'topic'" in capsys.readouterr().err
def test_run_json_crew_accepts_inputs_argument(monkeypatch, tmp_path: Path, capsys):
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("CREWAI_DMN", "True")
crew_path = tmp_path / "crew.jsonc"
crew_path.write_text("{}")
kickoff_calls: list[dict] = []
class FakeCrew:
name = "Demo"
agents = [_crew_with_placeholders("topic").agents[0]]
tasks: list = []
def kickoff(self, inputs):
kickoff_calls.append(inputs)
return "ok"
monkeypatch.setattr(
run_crew_module, "configured_project_json_crew", lambda: crew_path
)
monkeypatch.setattr(
run_crew_module, "_load_json_crew", lambda _path: (FakeCrew(), {})
)
assert run_crew_module._run_json_crew(inputs='{"topic":"AI"}') == "ok"
assert kickoff_calls == [{"topic": "AI"}]
def test_run_json_crew_in_project_env_forwards_inputs(monkeypatch, tmp_path: Path):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
monkeypatch.setattr(
run_crew_module, "_install_json_crew_dependencies_if_needed", lambda: None
)
monkeypatch.setattr(
run_crew_module, "build_env_with_all_tool_credentials", lambda: {}
)
captured_kwargs: list[dict] = []
monkeypatch.setattr(
run_crew_module.subprocess,
"run",
lambda command, **kwargs: captured_kwargs.append(kwargs),
)
run_crew_module._run_json_crew_in_project_env(
crew_path=tmp_path / "crew.jsonc", inputs='{"topic":"AI"}'
)
env = captured_kwargs[0]["env"]
assert env[run_crew_module._CREWAI_JSON_CREW_INPUTS_ENV] == '{"topic":"AI"}'
def test_run_json_crew_in_project_env_rejects_invalid_inputs_json(
monkeypatch, tmp_path: Path, capsys
):
monkeypatch.chdir(tmp_path)
(tmp_path / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
monkeypatch.setattr(
run_crew_module.subprocess,
"run",
lambda *a, **k: pytest.fail("subprocess must not run on invalid --inputs"),
)
with pytest.raises(SystemExit) as exc_info:
run_crew_module._run_json_crew_in_project_env(
crew_path=tmp_path / "crew.jsonc", inputs="not json"
)
assert exc_info.value.code == 1
assert "Invalid --inputs JSON" in capsys.readouterr().err
def test_run_crew_forwards_inputs_to_json_crew(monkeypatch):
monkeypatch.setattr(run_crew_module, "read_toml", lambda *a, **k: {})
monkeypatch.setattr(
run_crew_module,
"configured_project_json_crew",
lambda *a, **k: Path("crew.jsonc"),
)
called: dict = {}
monkeypatch.setattr(
run_crew_module,
"_run_json_crew_in_project_env",
lambda **kw: called.update(kw),
)
run_crew_module.run_crew(inputs='{"topic":"AI"}')
assert called == {
"trained_agents_file": None,
"crew_path": Path("crew.jsonc"),
"inputs": '{"topic":"AI"}',
}
def test_configured_project_json_crew_defers_to_declared_flow_type(
@@ -682,9 +855,9 @@ def test_configured_project_json_crew_ignores_missing_pyproject(
assert run_crew_module.configured_project_json_crew() is None
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".
def test_run_crew_inputs_rejected_for_classic_crew(monkeypatch):
# --inputs works for declarative flows and declarative (JSON) crews, but a
# classic crew takes its inputs from main.py, so it errors clearly.
monkeypatch.setattr(run_crew_module, "read_toml", lambda *a, **k: {})
monkeypatch.setattr(
run_crew_module, "configured_project_json_crew", lambda *a, **k: None
@@ -697,7 +870,10 @@ def test_run_crew_inputs_without_definition_rejected_for_non_flow(monkeypatch):
with pytest.raises(click.UsageError) as exc_info:
run_crew_module.run_crew(inputs='{"topic":"AI"}')
assert "--inputs is only supported for declarative flows" in exc_info.value.message
assert (
"--inputs is only supported for declarative flows and crews"
in exc_info.value.message
)
def test_run_crew_inputs_without_definition_resolves_configured_flow(monkeypatch):

3
lib/cli/tests/test_run_declarative_flow.py
View File

@@ -5,6 +5,7 @@ from pathlib import Path
import pytest
import crewai_cli.input_prompt as input_prompt_module
import crewai_cli.run_declarative_flow as run_declarative_flow_module
@@ -289,7 +290,7 @@ def test_prompts_for_missing_required_when_interactive(
prompted.append(text)
return "typed@example.com"
monkeypatch.setattr(run_declarative_flow_module.click, "prompt", fake_prompt)
monkeypatch.setattr(input_prompt_module.click, "prompt", fake_prompt)
run_declarative_flow_module.run_declarative_flow(str(path))