andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-09-22 02:46:39 +00:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

feat(cli): crewai eval evaluates the last traced run through AMP (#7649)
Some checks are pending
CodeQL Advanced / Analyze (actions) (push) Waiting to run
Details
CodeQL Advanced / Analyze (python) (push) Waiting to run
Details
Vulnerability Scan / Detect changes (push) Waiting to run
Details
Vulnerability Scan / pip-audit (push) Blocked by required conditions
Details

Browse Source 
* feat(cli): crewai eval evaluates the last traced run through AMP

`crewai eval` reads `.crewai/last_run.json` — the record crewAI writes
when a traced run's spans reach Wharf — and asks AMP to evaluate that
run: POST /crewai_plus/api/v1/tracing/evaluations with the execution id,
sending the saved `crewai login` when there is one and nothing otherwise.
AMP answers with an evaluation id and a URL; the command prints the URL,
opens it, waits for the verdict and prints it (goal gate and the four
grades), exit 1 only when the evaluation itself failed. `--run
EXECUTION_ID` evaluates another run.

With no traced run recorded it offers to turn tracing on for the project
(`CREWAI_TRACING_ENABLED=true` in .env, set_key so nothing else in the
file moves) and run the crew now with `crewai run`; without a terminal, or
declined, it prints the three steps instead. A run that leaves no record
behind is explained, never guessed at.

AMP's refusals are printed in its own words: a run that needs an account
(401 account_required), a refused credential (then `crewai login`), a run
AMP does not hold (404), rate limiting (429 with Retry-After), and any
other status with AMP's message. The two AMP calls live on the CLI's
PlusAPI subclass, so no crewai-core release is needed.

Who may evaluate what is AMP's decision, not the command's: an anonymous
run once without an account, then it needs one; a run traced while logged
in for that organization's members; a deployment execution for members
who may see its traces.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix(cli): crewai eval — guard the project, survive AMP blips, name the right subject, read the whole record

Review findings, each reproduced before the fix:

- The run-it-now offer wrote CREWAI_TRACING_ENABLED into ./.env before
  checking the directory is a crewAI project, then run_crew() died on a
  missing pyproject.toml with a traceback. Now: no pyproject.toml → one
  sentence, exit 1, nothing written.
- httpx errors (AMP unreachable, a timeout) surfaced as tracebacks. Now
  the start says "Could not reach AMP to start the evaluation: …"; while
  waiting, an unreachable AMP or a 5xx is retried up to POLL_RETRIES
  consecutive times, then reported with the URL — the evaluation keeps
  running server-side either way.
- The POST now carries a 120 s timeout (AMP reads the run's spans inside
  it), the poll 30 s.
- A 200 whose body has no known status (a non-dict, no status, a status
  outside queued/running/done/failed) polled forever. Now it stops with
  the status it saw and the URL.
- A 404 without a JSON message read "AMP holds no run <evaluation id>"
  while polling. _refused takes "run <id>" / "evaluation <id>" and says
  "AMP answered 404 for <subject>" — AMP's own message still wins.
- The record's amp_base_url was ignored; the CLI now evaluates the run at
  the AMP it was traced to. --run keeps the configured AMP.
- The post-run explanation names the third cause: a crewai older than the
  version that records the last run.

Tests for each, plus the previously untested paths: Ctrl-C exits 130, a
2xx without an id, a refusal mid-poll, DMN opens no browser, --run skips
the offer. 51 passed in lib/cli/tests (eval + plus_api).

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix(cli): crewai eval — the credential goes only to the configured AMP; an explicit yes before running the crew; a done answer needs a well-formed verdict

Review findings (CodeRabbit, the PR gate):

- The record's amp_base_url was handed to PlusAPI beside the saved login,
  so a modified .crewai/last_run.json could send the token to any origin.
  The client is now built from the configured AMP only (CREWAI_PLUS_URL,
  the saved settings, app.crewai.com); the project's .env is loaded first,
  as `crewai run` loads it, so the configured AMP is the one the run was
  traced to. A record naming another address gets a one-line note and no
  credential.
- The offer to turn tracing on and run the crew defaults to no and says
  the .env change stays; Enter no longer spends a crew run.
- A `done` answer whose verdict is missing or malformed (no gate, grades
  not an object, a grade not an int or null) is a protocol error, exit 1,
  instead of an INCONCLUSIVE line with exit 0 or an AttributeError.

Tests for each: a foreign origin in the record with a saved token, the
.env-loaded same-AMP case, seven malformed verdicts, the prompt's text and
default. 59 passed (eval + plus_api); ruff and mypy clean.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix(cli): crewai eval — Enter accepts the offer to run the crew (y/n, Y default; the prompt names both effects)

João's call (2026-09-20): the confirm is y/n with Y as the default. The
prompt still says tracing stays on in .env and that the crew runs now.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
João Moura
2026-09-21 15:34:04 -03:00
committed by GitHub
parent 6e1fa8cf2a
commit 0a3b8912b3
5 changed files with 956 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
lib/cli/src/crewai_cli/cli.py
View File

@@ -48,6 +48,12 @@ def run_crew(*args: Any, **kwargs: Any) -> Any:
return _run_crew(*args, **kwargs)
def eval_crew(*args: Any, **kwargs: Any) -> Any:
from crewai_cli.eval_crew import eval_crew as _eval_crew
return _eval_crew(*args, **kwargs)
if TYPE_CHECKING:
# mypy sees the real classes; at runtime the shims below defer the
# heavy imports until a command actually instantiates them.
@@ -674,6 +680,23 @@ def run(
)
@crewai.command(name="eval")
@click.option(
"--run",
"run_id",
type=str,
default=None,
metavar="EXECUTION_ID",
help=(
"Evaluate this traced run instead of the last one. The execution id "
"crewAI recorded for the run."
),
)
def eval_command(run_id: str | None) -> None:
"""Evaluate the last traced run through CrewAI AMP."""
eval_crew(run_id=run_id)
@crewai.command()
def update() -> None:
"""Update the pyproject.toml of the Crew project to use uv."""

377
lib/cli/src/crewai_cli/eval_crew.py Normal file
View File

@@ -0,0 +1,377 @@
"""`crewai eval`: evaluate the last traced run through CrewAI AMP.
crewAI records a traced run in `.crewai/last_run.json` when the run's spans
reach Wharf. This command reads that record (or takes `--run EXECUTION_ID`),
asks AMP to evaluate the run, prints and opens the URL AMP answers with,
waits for the verdict and prints it. With no traced run recorded it offers
to turn tracing on for the project and run the crew now.
Who may evaluate what is AMP's decision: an anonymous run once without an
account, then it needs one; a run traced while logged in for that
organization's members; a deployment execution for members who may see its
traces. The command sends the saved `crewai login` when there is one.
"""
from __future__ import annotations
import contextlib
from ipaddress import ip_address
import json
import os
from pathlib import Path
import sys
import time
from typing import Any
from urllib.parse import urlparse
import webbrowser
import click
from crewai_core.constants import DEFAULT_CREWAI_ENTERPRISE_URL
from crewai_core.settings import Settings
from dotenv import load_dotenv, set_key
import httpx
from rich.console import Console
from crewai_cli.authentication.token import AuthError, get_auth_token
from crewai_cli.plus_api import PlusAPI
from crewai_cli.utils import get_or_create_project_id, is_dmn_mode_enabled
console = Console()
LAST_RUN_FILE = Path(".crewai") / "last_run.json"
TRACING_ENV_VAR = "CREWAI_TRACING_ENABLED"
POLL_SECONDS = 3.0
POLL_RETRIES = 5 # consecutive unreachable / 5xx polls before giving up; the evaluation keeps running
FINISHED = {"done", "failed"}
STATUSES = {"queued", "running"} | FINISHED
def eval_crew(run_id: str | None = None) -> None:
"""Evaluate the last traced run of this project, or the run RUN_ID."""
get_or_create_project_id()
# Read before the project's .env is loaded, so a project cannot add itself.
trusted = _trusted_amp_origins()
_load_project_env()
record = read_last_run() or {}
execution_id = run_id or record.get("execution_id")
if execution_id is None:
execution_id = _run_now_or_explain()
record = read_last_run() or {}
client = _amp_client(trusted)
recorded_amp = str(record.get("amp_base_url") or "").rstrip("/")
if not run_id and recorded_amp and recorded_amp != client.base_url.rstrip("/"):
console.print(
f"The run was traced to {recorded_amp}; evaluating at the configured AMP {client.base_url}.",
style="yellow",
)
started = _start_evaluation(client, execution_id)
url = started.get("url")
console.print(f"Evaluating run [bold]{execution_id}[/bold]")
if url:
console.print(f"Follow it at [cyan underline]{url}[/cyan underline]")
_open(url)
finished = _wait(client, str(started["id"]), url)
_print_verdict(finished, url)
if finished.get("status") != "done":
raise SystemExit(1)
def _trusted_amp_origins() -> set[str]:
"""Where the saved login may be sent: the AMP this machine is configured for
(`crewai enterprise configure`), one already exported in this shell, and
crewAI's own."""
candidates = (
os.environ.get("CREWAI_PLUS_URL"),
Settings().enterprise_base_url,
DEFAULT_CREWAI_ENTERPRISE_URL,
)
return {origin for origin in map(_origin, candidates) if origin}
def _origin(url: str | None) -> str | None:
parsed = urlparse(str(url or ""))
return (
f"{parsed.scheme}://{parsed.netloc}".lower()
if parsed.scheme and parsed.netloc
else None
)
def _encrypted(origin: str | None) -> bool:
"""HTTPS, or plain HTTP to this machine — the rule `TraceGrantClient` already
applies to collector grants (localhost, its subdomains, loopback addresses)."""
parsed = urlparse(origin or "")
if parsed.scheme == "https":
return True
if parsed.scheme != "http":
return False
hostname = (parsed.hostname or "").rstrip(".")
if hostname == "localhost" or hostname.endswith(".localhost"):
return True
try:
return ip_address(hostname).is_loopback
except ValueError:
return False
def _amp_client(trusted: set[str]) -> PlusAPI:
"""The AMP to ask, and whether the saved login goes with it.
A project's `.env` may point `crewai eval` at another AMP — that is how a
self-hosted project is wired, and the run was traced there — so the request
follows it. The credential does not: it goes only to an AMP this machine is
logged in to, over a connection that encrypts it. Anywhere else the run is
read anonymously."""
client = PlusAPI(api_key=saved_login())
if client.api_key is None:
return client
origin = _origin(client.base_url)
if origin in trusted and _encrypted(origin):
return client
why = (
"is not an AMP this machine is logged in to"
if origin not in trusted
else "would carry the login over plain HTTP"
)
console.print(
f"Reading anonymously: {client.base_url} {why}. "
"Run `crewai enterprise configure <url>` to log in to it.",
style="yellow",
)
return PlusAPI()
def _load_project_env() -> None:
"""The project's .env, as `crewai run` loads it — so CREWAI_PLUS_URL here is the one the run used."""
env_file = Path.cwd() / ".env"
if env_file.is_file():
load_dotenv(env_file, override=True)
def read_last_run(directory: Path | None = None) -> dict[str, Any] | None:
"""The record crewAI wrote for the project's last traced run (execution_id,
tier, started_at, finished_at, amp_base_url), or None."""
path = (directory or Path.cwd()) / LAST_RUN_FILE
try:
loaded = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError):
return None
if not isinstance(loaded, dict) or not loaded.get("execution_id"):
return None
loaded["execution_id"] = str(loaded["execution_id"])
return loaded
def saved_login() -> str | None:
"""The `crewai login` token, or None: AMP then treats the caller as anonymous.
Only "not logged in" reads as anonymous. A credential that exists but cannot
be read — a rotated key, a half-written store, a directory `sudo` left owned
by root — is said out loud: reading anonymously instead would quietly spend
the run's one anonymous read and then refuse a user who believes they are
logged in."""
try:
return get_auth_token()
except AuthError:
return None
except Exception as error:
_fail(
f"Could not read the saved login ({type(error).__name__}: {error}). "
"Run `crewai login` again, or `crewai eval` will not know who you are."
)
return None
def _run_now_or_explain() -> str:
"""No traced run recorded here: offer to turn tracing on and run the crew now."""
if not Path("pyproject.toml").is_file():
_fail(
"No crewAI project here (no pyproject.toml). Run `crewai eval` from the project's "
"directory, or name a run: `crewai eval --run EXECUTION_ID`."
)
steps = (
"No traced run is recorded in this project. Turn tracing on and run the crew, "
f"then come back:\n 1. add {TRACING_ENV_VAR}=true to .env\n 2. crewai run\n 3. crewai eval"
)
if is_dmn_mode_enabled() or not sys.stdin.isatty():
console.print(steps, style="yellow")
raise SystemExit(1)
if not click.confirm(
f"No traced run is recorded in this project. Turn tracing on ({TRACING_ENV_VAR}=true "
"stays in .env) and run the crew now?",
default=True, # João, 2026-09-20: y/n with Y as the default — the prompt says what Enter does
):
console.print(steps, style="yellow")
raise SystemExit(0)
_enable_tracing()
from crewai_cli.run_crew import run_crew
run_crew()
record = read_last_run()
if record is None:
console.print(
"The run finished but no trace was recorded: the run may have failed, sharing the "
"trace was declined, or this project's crewai is older than the version that records "
f"the last run ({LAST_RUN_FILE}). Run the crew again and accept when asked, then `crewai eval`.",
style="bold red",
)
raise SystemExit(1)
return str(record["execution_id"])
def _enable_tracing() -> None:
"""`CREWAI_TRACING_ENABLED=true` in the project's .env, and in this process for the run about to start."""
env_file = Path.cwd() / ".env"
env_file.touch(exist_ok=True)
set_key(str(env_file), TRACING_ENV_VAR, "true", quote_mode="never")
os.environ[TRACING_ENV_VAR] = "true"
console.print(
f"Tracing is on for this project ({TRACING_ENV_VAR}=true in .env).",
style="green",
)
def _start_evaluation(client: PlusAPI, execution_id: str) -> dict[str, Any]:
try:
response = client.create_evaluation(execution_id)
except httpx.HTTPError as error:
_fail(f"Could not reach AMP to start the evaluation: {error}")
if response.status_code in (200, 202):
payload = _payload(response)
if payload and payload.get("id"):
return payload
_fail(f"AMP answered without an evaluation id ({response.status_code}).")
_refused(response, f"run {execution_id}")
raise AssertionError("unreachable")
def _wait(client: PlusAPI, evaluation_id: str, url: str | None) -> dict[str, Any]:
"""Poll until the evaluation is done or failed; Ctrl-C leaves it running."""
console.print("Waiting for the verdict…", style="dim")
where = f" at {url}" if url else ""
subject = f"evaluation {evaluation_id}"
misses = (
0 # AMP unreachable or answering 5xx: a blip is retried, a streak is reported
)
try:
while True:
try:
response = client.get_evaluation(evaluation_id)
except httpx.HTTPError as error:
misses += 1
if misses >= POLL_RETRIES:
_fail(
f"Could not reach AMP while waiting ({error}); the evaluation keeps running{where}."
)
time.sleep(POLL_SECONDS)
continue
if response.status_code >= 500:
misses += 1
if misses >= POLL_RETRIES:
_refused(response, subject)
time.sleep(POLL_SECONDS)
continue
if response.status_code != 200:
_refused(response, subject)
misses = 0
payload = _payload(response) or {}
status = payload.get("status")
if status == "done" and not _well_formed_verdict(payload.get("verdict")):
_fail(
f"AMP answered done without a verdict (protocol error); follow it{where or ' on AMP'}."
)
if status in FINISHED:
return payload
if status not in STATUSES:
_fail(
f"AMP answered without a known evaluation status ({status!r}); follow it{where or ' on AMP'}."
)
time.sleep(POLL_SECONDS)
except KeyboardInterrupt:
console.print(f"\nStill running{where}.", style="yellow")
raise SystemExit(130) from None
def _well_formed_verdict(verdict: Any) -> bool:
"""`{"gate": "<word>", "grades": {area: 1..5 | null}}` — anything else is a
protocol error. A grade is an exact integer in range: `True` is an `int` to
Python and 6 is not a grade, and neither may print as one."""
return (
isinstance(verdict, dict)
and isinstance(verdict.get("gate"), str)
and isinstance(verdict.get("grades"), dict)
and all(_a_grade(grade) for grade in verdict["grades"].values())
)
def _a_grade(grade: Any) -> bool:
return grade is None or (type(grade) is int and 1 <= grade <= 5)
def _print_verdict(finished: dict[str, Any], url: str | None) -> None:
if finished.get("status") != "done":
console.print(
f"Evaluation failed: {finished.get('error') or 'no reason given'}",
style="bold red",
)
return
verdict = finished["verdict"] # _wait let only a well-formed one through
gate = str(verdict["gate"]).upper()
style = {"PASSED": "bold green", "FAILED": "bold red"}.get(gate, "bold yellow")
grades = verdict.get("grades") or {}
parts = [
f"{area} {grades[area]}/5"
if grades.get(area) is not None
else f"{area} not measured"
for area in ("goal", "quality", "process", "cost")
]
console.print(f"Goal gate: [{style}]{gate}[/{style}] · " + " · ".join(parts))
if url:
console.print(f"Full report: {url}")
def _open(url: str) -> None:
if is_dmn_mode_enabled():
return
with contextlib.suppress(Exception): # no browser is not an error
webbrowser.open(url)
def _payload(response: httpx.Response) -> dict[str, Any] | None:
try:
loaded = response.json()
except ValueError:
return None
return loaded if isinstance(loaded, dict) else None
def _refused(response: httpx.Response, subject: str) -> None:
"""AMP's own words when it sent them, then exit 1. SUBJECT is "run <id>" or "evaluation <id>"."""
payload = _payload(response) or {}
message = str(payload.get("message") or "").strip()
error = str(payload.get("error") or "")
if response.status_code in (401, 403):
if error == "account_required" and message:
_fail(message)
_fail(
f"{message or 'AMP refused the credential'}. Log in with `crewai login` and try again."
)
if response.status_code == 404:
_fail(message or f"AMP answered 404 for {subject}.")
if response.status_code == 429:
retry = response.headers.get("Retry-After")
_fail(
f"{message or 'AMP is rate limiting this request'}{f' — retry after {retry}s' if retry else ''}."
)
_fail(f"AMP answered {response.status_code}{': ' + message if message else ''}.")
def _fail(message: str) -> None:
console.print(message, style="bold red")
raise SystemExit(1)

25
lib/cli/src/crewai_cli/plus_api.py
View File

@@ -18,9 +18,32 @@ class PlusAPI(_CorePlusAPI):
The ZIP deployment methods live here as well as in newer crewai-core
versions so editable CLI installs still work when an older crewai-core is
present in the runtime environment.
present in the runtime environment. The evaluation methods live here
because only the CLI calls them.
"""
EVALUATIONS_RESOURCE = f"{_CorePlusAPI.TRACING_RESOURCE}/evaluations"
# AMP reads the run's spans from Wharf inside the POST; a large run takes a while.
EVALUATION_START_TIMEOUT = 120.0
EVALUATION_POLL_TIMEOUT = 30.0
def create_evaluation(self, execution_id: str) -> httpx.Response:
"""Ask AMP to evaluate the traced run EXECUTION_ID (crewai eval)."""
return self._make_request(
"POST",
self.EVALUATIONS_RESOURCE,
json={"execution_id": execution_id},
timeout=self.EVALUATION_START_TIMEOUT,
)
def get_evaluation(self, evaluation_id: str) -> httpx.Response:
"""The evaluation's status and, once done, its verdict."""
return self._make_request(
"GET",
f"{self.EVALUATIONS_RESOURCE}/{evaluation_id}",
timeout=self.EVALUATION_POLL_TIMEOUT,
)
def _make_multipart_request(
self,
method: HttpMethod,

505
lib/cli/tests/test_eval_crew.py Normal file
View File

@@ -0,0 +1,505 @@
"""`crewai eval`: the last traced run, evaluated through AMP."""
from __future__ import annotations
import json
import os
from pathlib import Path
from types import SimpleNamespace
from click.testing import CliRunner
import httpx
import pytest
from rich.console import Console
from crewai_cli import eval_crew as eval_module
from crewai_cli.cli import eval_command
EXECUTION_ID = "6f31fe1a-20bd-4bfe-a011-25d6b9341f62"
URL = "https://evolve.crewai.test/e/ev-1"
class FakeAMP:
"""A PlusAPI double: scripted answers, calls recorded."""
def __init__(self, create=None, statuses=None):
self.create = create if create is not None else httpx.Response(
202, json={"id": "ev-1", "url": URL, "status": "queued"}
)
self.statuses = list(statuses or [])
self.calls: list[tuple] = []
self.api_key = None
def create_evaluation(self, execution_id):
self.calls.append(("create", execution_id))
return self.create
def get_evaluation(self, evaluation_id):
self.calls.append(("get", evaluation_id))
return self.statuses.pop(0) if self.statuses else httpx.Response(200, json={"id": evaluation_id, "status": "running"})
def done(gate="passed", grades=None):
return httpx.Response(200, json={
"id": "ev-1", "status": "done", "url": URL,
"verdict": {"gate": gate, "grades": grades if grades is not None else {"goal": 5, "quality": 4, "process": 5, "cost": None}},
})
@pytest.fixture
def project(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(eval_module, "console", Console(width=240)) # one sentence per line in the captured output
monkeypatch.setattr(eval_module, "get_or_create_project_id", lambda: None)
monkeypatch.setattr(eval_module, "saved_login", lambda: "login-token")
monkeypatch.setattr(eval_module.time, "sleep", lambda seconds: None)
monkeypatch.setattr(eval_module, "is_dmn_mode_enabled", lambda: False)
# This machine is logged in to https://amp.test (`crewai enterprise configure`).
monkeypatch.setattr(eval_module, "Settings", lambda: SimpleNamespace(enterprise_base_url="https://amp.test"))
monkeypatch.delenv("CREWAI_PLUS_URL", raising=False)
opened: list[str] = []
monkeypatch.setattr(eval_module.webbrowser, "open", lambda url: opened.append(url) or True)
return tmp_path, opened
def record_last_run(directory: Path, execution_id: str = EXECUTION_ID, **fields) -> None:
(directory / ".crewai").mkdir(exist_ok=True)
record = {"execution_id": execution_id, "tier": "ephemeral", "amp_base_url": "https://amp.test", **fields}
(directory / ".crewai" / "last_run.json").write_text(json.dumps(record))
def install(monkeypatch, amp: FakeAMP, configured_amp: str = "https://amp.test") -> FakeAMP:
def build(api_key=None, base_url=None):
# PlusAPI's own resolution: explicit, then CREWAI_PLUS_URL, then the saved settings.
amp.api_key = api_key
amp.base_url = base_url or os.environ.get("CREWAI_PLUS_URL") or configured_amp
return amp
monkeypatch.setattr(eval_module, "PlusAPI", build)
return amp
def test_the_last_run_is_evaluated_the_url_opened_and_the_verdict_printed(project, monkeypatch, capsys):
directory, opened = project
record_last_run(directory)
amp = install(monkeypatch, FakeAMP(statuses=[httpx.Response(200, json={"id": "ev-1", "status": "running"}), done()]))
eval_module.eval_crew()
out = capsys.readouterr().out
assert amp.api_key == "login-token"
assert amp.calls == [("create", EXECUTION_ID), ("get", "ev-1"), ("get", "ev-1")]
assert opened == [URL]
assert EXECUTION_ID in out and URL in out
assert "Goal gate: PASSED" in out and "goal 5/5" in out and "cost not measured" in out
def test_run_names_another_execution_and_an_anonymous_caller_sends_no_token(project, monkeypatch, capsys):
directory, _ = project
record_last_run(directory)
monkeypatch.setattr(eval_module, "saved_login", lambda: None)
amp = install(monkeypatch, FakeAMP(statuses=[done("failed")]))
eval_module.eval_crew(run_id="other-run")
assert amp.api_key is None
assert amp.calls[0] == ("create", "other-run")
assert "Goal gate: FAILED" in capsys.readouterr().out
def test_a_failed_evaluation_exits_one_with_amps_reason(project, monkeypatch, capsys):
directory, _ = project
record_last_run(directory)
install(monkeypatch, FakeAMP(statuses=[httpx.Response(200, json={"id": "ev-1", "status": "failed", "error": "the judge was unreachable"})]))
with pytest.raises(SystemExit) as exit_:
eval_module.eval_crew()
assert exit_.value.code == 1
assert "Evaluation failed: the judge was unreachable" in capsys.readouterr().out
@pytest.mark.parametrize(
("response", "expected"),
[
(httpx.Response(401, json={"error": "account_required", "message": "Execution x was already read once without an account. Log in with `crewai login`, or create an account, to read it again."}),
"already read once without an account"),
(httpx.Response(401, json={"error": "bad_credentials", "message": "Bad credentials"}), "Bad credentials. Log in with `crewai login`"),
(httpx.Response(404, json={"error": "trace_not_found", "message": "No spans recorded for execution 6f31"}), "No spans recorded for execution 6f31"),
(httpx.Response(404, text="<html>Page not found</html>"), f"AMP answered 404 for run {EXECUTION_ID}."),
(httpx.Response(202, json={"url": URL}), "AMP answered without an evaluation id (202)."),
(httpx.Response(429, json={"error": "rate_limit_exceeded", "message": "Too many requests"}, headers={"Retry-After": "60"}), "Too many requests — retry after 60s"),
(httpx.Response(503, json={"error": "service_unavailable", "message": "Wharf could not list the spans"}), "AMP answered 503: Wharf could not list the spans"),
(httpx.Response(500, text="boom"), "AMP answered 500."),
],
)
def test_amps_refusals_are_printed_in_its_words_and_exit_one(project, monkeypatch, capsys, response, expected):
directory, opened = project
record_last_run(directory)
install(monkeypatch, FakeAMP(create=response))
with pytest.raises(SystemExit) as exit_:
eval_module.eval_crew()
assert exit_.value.code == 1
assert expected in capsys.readouterr().out
assert opened == []
def test_the_credential_goes_only_to_the_configured_amp_never_to_an_address_off_the_record(project, monkeypatch, capsys):
directory, _ = project
record_last_run(directory, amp_base_url="https://evil.example/steal")
amp = install(monkeypatch, FakeAMP(statuses=[done()]), configured_amp="https://app.crewai.com")
eval_module.eval_crew()
assert amp.api_key == "login-token" and amp.base_url == "https://app.crewai.com" # PlusAPI got no base_url
out = capsys.readouterr().out
assert "The run was traced to https://evil.example/steal; evaluating at the configured AMP https://app.crewai.com." in out
# The project's .env is what `crewai run` traced with, so it is loaded first. This machine is
# logged in to that AMP, so the credential goes with the request and nothing is remarked on.
(directory / ".env").write_text("CREWAI_PLUS_URL=https://amp.test\n")
record_last_run(directory, amp_base_url="https://amp.test/")
amp = install(monkeypatch, FakeAMP(statuses=[done()]))
eval_module.eval_crew()
assert eval_module.os.environ["CREWAI_PLUS_URL"] == "https://amp.test"
assert amp.api_key == "login-token" and amp.base_url == "https://amp.test"
out = capsys.readouterr().out
assert "was traced to" not in out and "Reading anonymously" not in out
def test_a_project_may_point_at_another_amp_but_never_gets_the_saved_login(project, monkeypatch, capsys):
"""A .env can send the request elsewhere — that is how a self-hosted project is wired —
but the token goes only to an AMP this machine is logged in to."""
directory, _ = project
(directory / ".env").write_text("CREWAI_PLUS_URL=https://evil.example\n")
record_last_run(directory, amp_base_url="https://evil.example")
amp = install(monkeypatch, FakeAMP(statuses=[done()]))
eval_module.eval_crew() # still works: AMP reads it anonymously
assert amp.base_url == "https://evil.example" # the request follows the project
assert amp.api_key is None # the credential does not
out = capsys.readouterr().out
assert "Reading anonymously: https://evil.example is not an AMP this machine is logged in to." in out
assert "crewai enterprise configure" in out
def test_a_trusted_amp_over_plain_http_still_gets_no_credential(project, monkeypatch, capsys):
"""A cleartext connection is not a place to put a bearer token, trusted or not."""
directory, _ = project
monkeypatch.setenv("CREWAI_PLUS_URL", "http://amp.internal") # exported, so it IS trusted
record_last_run(directory, amp_base_url="http://amp.internal")
amp = install(monkeypatch, FakeAMP(statuses=[done()]))
eval_module.eval_crew()
assert amp.base_url == "http://amp.internal" and amp.api_key is None
assert "would carry the login over plain HTTP" in capsys.readouterr().out
def test_plain_http_to_this_machine_is_fine_for_local_development(project, monkeypatch, capsys):
directory, _ = project
monkeypatch.setenv("CREWAI_PLUS_URL", "http://localhost:3000")
record_last_run(directory, amp_base_url="http://localhost:3000")
amp = install(monkeypatch, FakeAMP(statuses=[done()]))
eval_module.eval_crew()
assert amp.api_key == "login-token"
assert "Reading anonymously" not in capsys.readouterr().out
@pytest.mark.parametrize(
("origin", "encrypted"),
[
("https://app.crewai.com", True),
("http://localhost:3000", True),
("http://127.0.0.1:8000", True),
("http://[::1]:8000", True),
("http://amp.localhost", True),
("http://amp.internal", False),
("http://169.254.169.254", False),
("ftp://amp.test", False),
(None, False),
],
)
def test_which_connections_may_carry_the_login(origin, encrypted):
assert eval_module._encrypted(origin) is encrypted
def test_an_amp_exported_in_this_shell_is_trusted(project, monkeypatch, capsys):
directory, _ = project
monkeypatch.setenv("CREWAI_PLUS_URL", "https://shell.amp.test") # exported before the project is read
record_last_run(directory, amp_base_url="https://shell.amp.test")
amp = install(monkeypatch, FakeAMP(statuses=[done()]))
eval_module.eval_crew()
assert amp.base_url == "https://shell.amp.test" and amp.api_key == "login-token"
assert "Reading anonymously" not in capsys.readouterr().out
@pytest.mark.parametrize(
("url", "origin"),
[
("https://amp.test", "https://amp.test"),
("https://AMP.Test/crewai_plus/", "https://amp.test"),
("http://localhost:8000/x", "http://localhost:8000"),
("app.crewai.com", None), # no scheme: not an origin, never trusted
("", None),
(None, None),
],
)
def test_an_origin_is_scheme_and_host_only(url, origin):
assert eval_module._origin(url) == origin
@pytest.mark.parametrize(
"verdict",
[
None,
"passed",
[],
{"gate": "passed"},
{"gate": None, "grades": {}},
{"gate": "passed", "grades": "5/5"},
{"gate": "passed", "grades": {"goal": "five"}},
{"gate": "passed", "grades": {"goal": True}}, # a bool is an int to Python, never a grade
{"gate": "passed", "grades": {"goal": 6}}, # out of the 1..5 range
{"gate": "passed", "grades": {"goal": 0}},
{"gate": "passed", "grades": {"goal": 4.5}},
],
)
def test_a_done_answer_without_a_well_formed_verdict_is_a_protocol_error(project, monkeypatch, capsys, verdict):
directory, _ = project
record_last_run(directory)
body = {"id": "ev-1", "status": "done", "url": URL}
if verdict is not None:
body["verdict"] = verdict
install(monkeypatch, FakeAMP(statuses=[httpx.Response(200, json=body)]))
with pytest.raises(SystemExit) as exit_:
eval_module.eval_crew()
assert exit_.value.code == 1
assert f"AMP answered done without a verdict (protocol error); follow it at {URL}." in capsys.readouterr().out
def test_amp_unreachable_at_the_start_is_a_sentence_not_a_traceback(project, monkeypatch, capsys):
directory, _ = project
record_last_run(directory)
amp = install(monkeypatch, FakeAMP())
monkeypatch.setattr(amp, "create_evaluation", lambda execution_id: (_ for _ in ()).throw(httpx.ConnectError("connection refused")))
with pytest.raises(SystemExit) as exit_:
eval_module.eval_crew()
assert exit_.value.code == 1
assert "Could not reach AMP to start the evaluation: connection refused" in capsys.readouterr().out
def test_while_waiting_a_blip_is_retried_and_a_streak_is_reported(project, monkeypatch, capsys):
directory, _ = project
record_last_run(directory)
amp = install(monkeypatch, FakeAMP(statuses=[httpx.Response(502), httpx.Response(503, json={"error": "service_unavailable", "message": "crew-optimize is down"}), done()]))
eval_module.eval_crew() # two bad polls, then the verdict
assert "Goal gate: PASSED" in capsys.readouterr().out
assert amp.calls.count(("get", "ev-1")) == 3
record_last_run(directory)
amp = install(monkeypatch, FakeAMP(statuses=[httpx.Response(503, json={"error": "service_unavailable", "message": "crew-optimize is down"})] * eval_module.POLL_RETRIES))
with pytest.raises(SystemExit) as exit_:
eval_module.eval_crew()
assert exit_.value.code == 1
assert "AMP answered 503: crew-optimize is down" in capsys.readouterr().out
assert amp.calls.count(("get", "ev-1")) == eval_module.POLL_RETRIES
amp = install(monkeypatch, FakeAMP())
monkeypatch.setattr(amp, "get_evaluation", lambda evaluation_id: (_ for _ in ()).throw(httpx.ReadTimeout("timed out")))
with pytest.raises(SystemExit) as exit_:
eval_module.eval_crew()
assert exit_.value.code == 1
assert f"Could not reach AMP while waiting (timed out); the evaluation keeps running at {URL}." in capsys.readouterr().out
def test_a_refusal_mid_poll_names_the_evaluation_and_an_unknown_status_stops_the_wait(project, monkeypatch, capsys):
directory, _ = project
record_last_run(directory)
install(monkeypatch, FakeAMP(statuses=[httpx.Response(404, text="gone")]))
with pytest.raises(SystemExit) as exit_:
eval_module.eval_crew()
assert exit_.value.code == 1 and "AMP answered 404 for evaluation ev-1." in capsys.readouterr().out
for odd in (httpx.Response(200, json={"id": "ev-1", "status": "cancelled"}), httpx.Response(200, json=[]), httpx.Response(200, text="<html>")):
install(monkeypatch, FakeAMP(statuses=[httpx.Response(200, json={"id": "ev-1", "status": "queued"}), odd]))
with pytest.raises(SystemExit) as exit_:
eval_module.eval_crew()
assert exit_.value.code == 1
assert f"AMP answered without a known evaluation status" in capsys.readouterr().out
def test_ctrl_c_leaves_the_evaluation_running_and_exits_130(project, monkeypatch, capsys):
directory, _ = project
record_last_run(directory)
amp = install(monkeypatch, FakeAMP())
monkeypatch.setattr(amp, "get_evaluation", lambda evaluation_id: (_ for _ in ()).throw(KeyboardInterrupt()))
with pytest.raises(SystemExit) as exit_:
eval_module.eval_crew()
assert exit_.value.code == 130
assert f"Still running at {URL}." in capsys.readouterr().out
def test_dmn_mode_prints_the_url_but_opens_no_browser(project, monkeypatch, capsys):
directory, opened = project
record_last_run(directory)
monkeypatch.setattr(eval_module, "is_dmn_mode_enabled", lambda: True)
install(monkeypatch, FakeAMP(statuses=[done()]))
eval_module.eval_crew()
assert opened == [] and URL in capsys.readouterr().out
def test_run_skips_the_offer_when_nothing_is_recorded(project, monkeypatch, capsys):
monkeypatch.setattr(eval_module.click, "confirm", lambda *args, **kwargs: pytest.fail("no offer with --run"))
amp = install(monkeypatch, FakeAMP(statuses=[done()]))
eval_module.eval_crew(run_id="named-run")
assert amp.calls[0] == ("create", "named-run")
def test_outside_a_crewai_project_nothing_is_written_and_it_says_so(project, monkeypatch, capsys):
directory, _ = project
monkeypatch.setattr(eval_module.sys.stdin, "isatty", lambda: True)
monkeypatch.setattr(eval_module.click, "confirm", lambda *args, **kwargs: pytest.fail("no offer outside a project"))
amp = install(monkeypatch, FakeAMP())
with pytest.raises(SystemExit) as exit_:
eval_module.eval_crew()
assert exit_.value.code == 1
assert "No crewAI project here (no pyproject.toml)" in capsys.readouterr().out
assert not (directory / ".env").exists() and amp.calls == []
def test_without_a_traced_run_and_no_terminal_it_explains_and_exits(project, monkeypatch, capsys):
(project[0] / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
monkeypatch.setattr(eval_module, "is_dmn_mode_enabled", lambda: True)
amp = install(monkeypatch, FakeAMP())
with pytest.raises(SystemExit) as exit_:
eval_module.eval_crew()
assert exit_.value.code == 1
out = capsys.readouterr().out
assert "No traced run is recorded" in out and "CREWAI_TRACING_ENABLED=true" in out and "crewai run" in out
assert amp.calls == []
def test_without_a_traced_run_it_offers_to_turn_tracing_on_and_run_the_crew(project, monkeypatch, capsys):
directory, _ = project
(directory / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
monkeypatch.setattr(eval_module.sys.stdin, "isatty", lambda: True)
prompts: list[tuple] = []
monkeypatch.setattr(eval_module.click, "confirm", lambda text, **kwargs: prompts.append((text, kwargs)) or True)
ran: list[str] = []
def fake_run_crew() -> None:
ran.append("run")
assert eval_module.os.environ.get("CREWAI_TRACING_ENABLED") == "true"
record_last_run(directory, "fresh-run")
import crewai_cli.run_crew as run_crew_module
monkeypatch.setattr(run_crew_module, "run_crew", fake_run_crew)
monkeypatch.delenv("CREWAI_TRACING_ENABLED", raising=False)
amp = install(monkeypatch, FakeAMP(statuses=[done()]))
eval_module.eval_crew()
assert ran == ["run"]
text, kwargs = prompts[0]
assert "CREWAI_TRACING_ENABLED=true stays in .env" in text and kwargs == {"default": True} # Enter is yes (João's call); the prompt names both effects
assert "CREWAI_TRACING_ENABLED=true" in (directory / ".env").read_text()
assert amp.calls[0] == ("create", "fresh-run")
assert "Tracing is on for this project" in capsys.readouterr().out
def test_declining_the_offer_exits_cleanly_with_the_steps(project, monkeypatch, capsys):
(project[0] / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
monkeypatch.setattr(eval_module.sys.stdin, "isatty", lambda: True)
monkeypatch.setattr(eval_module.click, "confirm", lambda *args, **kwargs: False)
amp = install(monkeypatch, FakeAMP())
with pytest.raises(SystemExit) as exit_:
eval_module.eval_crew()
assert exit_.value.code == 0
assert "crewai eval" in capsys.readouterr().out and amp.calls == []
def test_a_run_that_leaves_no_trace_behind_is_explained(project, monkeypatch, capsys):
(project[0] / "pyproject.toml").write_text("[project]\nname = 'demo'\n")
monkeypatch.setattr(eval_module.sys.stdin, "isatty", lambda: True)
monkeypatch.setattr(eval_module.click, "confirm", lambda *args, **kwargs: True)
import crewai_cli.run_crew as run_crew_module
monkeypatch.setattr(run_crew_module, "run_crew", lambda: None)
install(monkeypatch, FakeAMP())
with pytest.raises(SystemExit) as exit_:
eval_module.eval_crew()
assert exit_.value.code == 1
out = capsys.readouterr().out
assert "no trace was recorded" in out and "older than the version that records the last run" in out
def test_the_cli_command_maps_to_the_implementation(monkeypatch):
calls = []
monkeypatch.setattr("crewai_cli.cli.eval_crew", lambda **kwargs: calls.append(kwargs))
runner = CliRunner()
assert runner.invoke(eval_command, []).exit_code == 0
assert runner.invoke(eval_command, ["--run", EXECUTION_ID]).exit_code == 0
assert calls == [{"run_id": None}, {"run_id": EXECUTION_ID}]
assert "Evaluate the last traced run" in runner.invoke(eval_command, ["--help"]).output
def test_only_a_missing_login_reads_as_anonymous(monkeypatch, capsys):
"""An unreadable credential store is not "anonymous": it is said out loud."""
from crewai_cli.authentication.token import AuthError
monkeypatch.setattr(eval_module, "get_auth_token", lambda: (_ for _ in ()).throw(AuthError("No token found")))
assert eval_module.saved_login() is None # not logged in: AMP treats the caller as anonymous
monkeypatch.setattr(eval_module, "get_auth_token", lambda: "login-token")
assert eval_module.saved_login() == "login-token"
for broken in (OSError(13, "Permission denied"), ValueError("Fernet key must be 32 url-safe base64-encoded bytes.")):
monkeypatch.setattr(eval_module, "get_auth_token", lambda error=broken: (_ for _ in ()).throw(error))
with pytest.raises(SystemExit) as exit_:
eval_module.saved_login()
assert exit_.value.code == 1
out = capsys.readouterr().out
assert "Could not read the saved login" in out and type(broken).__name__ in out and "crewai login" in out
def test_read_last_run_reads_the_record_crewai_writes(tmp_path):
assert eval_module.read_last_run(tmp_path) is None
(tmp_path / ".crewai").mkdir()
(tmp_path / ".crewai" / "last_run.json").write_text("not json")
assert eval_module.read_last_run(tmp_path) is None
(tmp_path / ".crewai" / "last_run.json").write_text(json.dumps({"tier": "ephemeral"}))
assert eval_module.read_last_run(tmp_path) is None
record_last_run(tmp_path)
record = eval_module.read_last_run(tmp_path)
assert record is not None and record["execution_id"] == EXECUTION_ID and record["amp_base_url"] == "https://amp.test"

27
lib/cli/tests/test_plus_api.py
View File

@@ -18,6 +18,33 @@ class TestPlusAPI(unittest.TestCase):
self.assertIn("CrewAI-CLI/", self.api.headers["User-Agent"])
self.assertTrue(self.api.headers["X-Crewai-Version"])
@patch("crewai_core.plus_api.PlusAPI._make_request")
def test_create_evaluation(self, mock_make_request):
mock_response = MagicMock()
mock_make_request.return_value = mock_response
response = self.api.create_evaluation("6f31fe1a-20bd-4bfe-a011-25d6b9341f62")
mock_make_request.assert_called_once_with(
"POST",
"/crewai_plus/api/v1/tracing/evaluations",
json={"execution_id": "6f31fe1a-20bd-4bfe-a011-25d6b9341f62"},
timeout=120.0, # AMP reads the run's spans inside this request
)
self.assertEqual(response, mock_response)
@patch("crewai_core.plus_api.PlusAPI._make_request")
def test_get_evaluation(self, mock_make_request):
mock_response = MagicMock()
mock_make_request.return_value = mock_response
response = self.api.get_evaluation("ev-1")
mock_make_request.assert_called_once_with(
"GET", "/crewai_plus/api/v1/tracing/evaluations/ev-1", timeout=30.0
)
self.assertEqual(response, mock_response)
@patch("crewai_core.plus_api.PlusAPI._make_request")
def test_login_to_tool_repository(self, mock_make_request):
mock_response = MagicMock()