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

Merge branch 'main' into joaomdmoura/run-inputs-definition-resolution

Browse Source
This commit is contained in:
João Moura
2026-07-07 02:17:18 -03:00
committed by GitHub
parent 67d9fcdcfc 799ab0f548
commit 792ca8ec75
16 changed files with 1721 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
lib/cli/src/crewai_cli/create_json_crew.py
View File

@@ -14,6 +14,7 @@ from rich.text import Text
from crewai_cli.constants import ENV_VARS
from crewai_cli.git import initialize_if_git_available
from crewai_cli.model_catalog import get_provider_models
from crewai_cli.tui_picker import pick_many, pick_one
from crewai_cli.utils import (
enable_prompt_line_editing,
@@ -42,41 +43,50 @@ _PROVIDERS: list[tuple[str, str]] = [
("watson", "IBM watsonx"),
]
# Curated offline fallback / label source. The picker prefers models pulled
# live from the vendor's own API via ``model_catalog.get_provider_models``;
# this list is the hand-verified backstop used when no API key is available.
# Keep entries to real, current model ids — last verified against each vendor's
# official model docs on 2026-07-05.
_PROVIDER_MODELS: dict[str, list[tuple[str, str]]] = {
"openai": [
("gpt-5.5", "GPT-5.5"),
("gpt-5.5-pro", "GPT-5.5 Pro"),
("gpt-5.4", "GPT-5.4"),
("o4-mini", "o4-mini"),
("gpt-5.4-mini", "GPT-5.4 Mini"),
("gpt-5.2", "GPT-5.2"),
("gpt-4.1", "GPT-4.1"),
("gpt-4.1-mini", "GPT-4.1 Mini"),
],
"anthropic": [
("claude-opus-4-6", "Claude Opus 4.6"),
("claude-fable-5", "Claude Fable 5"),
("claude-opus-4-8", "Claude Opus 4.8"),
("claude-sonnet-5", "Claude Sonnet 5"),
("claude-opus-4-7", "Claude Opus 4.7"),
("claude-haiku-4-5", "Claude Haiku 4.5"),
("claude-sonnet-4-6", "Claude Sonnet 4.6"),
("claude-haiku-4-5-20251001", "Claude Haiku 4.5"),
("claude-3-7-sonnet-20250219", "Claude 3.7 Sonnet"),
("claude-3-5-sonnet-20241022", "Claude 3.5 Sonnet"),
],
"gemini": [
("gemini-3-pro-preview", "Gemini 3 Pro (preview)"),
("gemini-2.5-pro-exp-03-25", "Gemini 2.5 Pro"),
("gemini-2.5-flash-preview-04-17", "Gemini 2.5 Flash"),
("gemini-2.0-flash-001", "Gemini 2.0 Flash"),
("gemini-1.5-pro", "Gemini 1.5 Pro"),
("gemini-3.5-flash", "Gemini 3.5 Flash"),
("gemini-3.1-pro-preview", "Gemini 3.1 Pro (preview)"),
("gemini-3-flash-preview", "Gemini 3 Flash (preview)"),
("gemini-2.5-pro", "Gemini 2.5 Pro"),
("gemini-2.5-flash", "Gemini 2.5 Flash"),
("gemini-2.5-flash-lite", "Gemini 2.5 Flash Lite"),
],
"groq": [
("meta-llama/llama-4-maverick-17b-128e-instruct", "Llama 4 Maverick"),
("meta-llama/llama-4-scout-17b-16e-instruct", "Llama 4 Scout"),
("openai/gpt-oss-120b", "GPT-OSS 120B"),
("qwen/qwen3-32b", "Qwen3 32B"),
("moonshotai/kimi-k2-instruct-0905", "Kimi K2"),
("llama-3.3-70b-versatile", "Llama 3.3 70B"),
("llama-3.1-70b-versatile", "Llama 3.1 70B"),
("llama-3.1-8b-instant", "Llama 3.1 8B"),
("deepseek-r1-distill-llama-70b", "DeepSeek R1 70B"),
("mixtral-8x7b-32768", "Mixtral 8x7B"),
],
"ollama": [
("llama3.3", "Llama 3.3"),
("llama3.1", "Llama 3.1"),
("qwen3", "Qwen 3"),
("deepseek-r1", "DeepSeek R1"),
("qwen2.5", "Qwen 2.5"),
("gpt-oss", "GPT-OSS"),
("gemma3", "Gemma 3"),
("mistral", "Mistral"),
],
}
@@ -758,7 +768,9 @@ def _select_model() -> str:
provider_key, provider_name = _PROVIDERS[p_idx]
click.secho(f" → {provider_name}", fg="green")
models = _PROVIDER_MODELS.get(provider_key, [])
# Prefer the latest models pulled live from the vendor / LiteLLM; the
# curated ``_PROVIDER_MODELS`` entry is the offline fallback and label source.
models = get_provider_models(provider_key, _PROVIDER_MODELS.get(provider_key, []))
if not models:
custom = click.prompt(
click.style(f" Enter model name for {provider_key}/", fg="cyan"),

657
lib/cli/src/crewai_cli/model_catalog.py Normal file
View File

@@ -0,0 +1,657 @@
"""Dynamic model catalog for the crew-creation wizard.
Resolves the models to offer for a given provider using a three-tier strategy:
1. **Vendor API** - when the provider's API key is already present in the
environment, query the vendor's own model-listing endpoint. This is the only
source that reliably reflects the *latest* models (real release dates /
display names, straight from the vendor).
2. **Curated hardcoded fallback** - the hand-verified list baked into the
wizard, used when no API key is available. Authoritative but frozen, so it is
refreshed periodically.
3. **LiteLLM feed** - the community ``model_prices_and_context_window.json`` the
CLI already caches. Only used for providers with *no* curated list: the feed
lags real releases badly (it can miss a vendor's newest models entirely), so
it must never preempt the curated fallback.
Every tier is best-effort: any network error, timeout, missing key, or empty
result quietly falls through to the next tier, and the caller's hardcoded list
is always the final backstop. The picker never blocks for long — network calls
use a short timeout and successful results are cached.
"""
from __future__ import annotations
from collections.abc import Callable
import contextlib
import json
import os
from pathlib import Path
import re
import time
from typing import Any
import certifi
import httpx
from crewai_cli.constants import JSON_URL
# ── Tunables ─────────────────────────────────────────────────────
#: How many models to surface per provider.
MAX_MODELS = 8
#: Timeout (seconds) for any network call made while resolving models.
_TIMEOUT = 6.0
#: How long a resolved (dynamic) catalog stays fresh before we refetch.
_CATALOG_TTL = 6 * 3600
#: How long a fallback result is cached after a failed/empty fetch. Short, so a
#: newly-added API key takes effect soon, but long enough to spare the picker a
#: repeated timeout-prone network attempt on every call within one session.
_NEGATIVE_TTL = 300
#: How long the shared LiteLLM feed cache stays fresh.
_LITELLM_TTL = 24 * 3600
#: Env vars that may hold each provider's API key, in priority order. A
#: provider with an empty tuple (e.g. local Ollama) needs no key. Gemini accepts
#: either name, matching crewai's own Gemini provider.
_PROVIDER_KEY_ENV: dict[str, tuple[str, ...]] = {
"openai": ("OPENAI_API_KEY",),
"anthropic": ("ANTHROPIC_API_KEY",),
"gemini": ("GEMINI_API_KEY", "GOOGLE_API_KEY"),
"groq": ("GROQ_API_KEY",),
"cerebras": ("CEREBRAS_API_KEY",),
"ollama": (),
}
def _provider_api_key(provider_key: str) -> str | None:
"""First non-empty API key found among the provider's env vars."""
for env in _PROVIDER_KEY_ENV.get(provider_key, ()):
value = os.environ.get(env)
if value:
return value
return None
# Substrings that mark a model id as *not* a chat/completion model. Used to
# filter noisy OpenAI-compatible ``/models`` listings.
_NON_CHAT_MARKERS = (
"embedding",
"embed",
"whisper",
"tts",
"audio",
"transcribe",
"realtime",
"dall-e",
"dalle",
"image",
"moderation",
"similarity",
"-edit",
"davinci-002",
"babbage-002",
"computer-use",
"guard",
)
_ACRONYMS = {
"gpt": "GPT",
"ai": "AI",
"nim": "NIM",
"llm": "LLM",
"hd": "HD",
"us": "US",
"eu": "EU",
"oss": "OSS",
"it": "IT",
}
# Tokens with non-title-case brand capitalization.
_BRAND_TOKENS = {
"deepseek": "DeepSeek",
"chatgpt": "ChatGPT",
"qwq": "QwQ",
}
# ── Public API ───────────────────────────────────────────────────
def get_provider_models(
provider_key: str, fallback: list[tuple[str, str]]
) -> list[tuple[str, str]]:
"""Return ``(model_id, label)`` pairs for ``provider_key``, newest first.
Tries the vendor API (if a key is in the environment) first, since it is the
only reliably-fresh source. When no key is available it returns the curated
``fallback`` verbatim — the LiteLLM feed is consulted **only** for providers
with no curated list, because the feed lags real releases and would
otherwise surface a staler list than the hand-verified fallback. Never
raises: any failure degrades to the next tier.
Args:
provider_key: Short provider identifier, e.g. ``"anthropic"``.
fallback: Curated ``(model_id, label)`` pairs to use as the backstop and
to source friendly labels for known models.
Returns:
Up to :data:`MAX_MODELS` ``(model_id, label)`` pairs. Falls back to
``fallback`` verbatim when no fresher list can be resolved.
"""
cached = _read_catalog_cache(provider_key)
if cached is not None:
return cached
label_map = {model_id: label for model_id, label in fallback}
# A non-None vendor result is authoritative — even when empty (e.g. a
# reachable Ollama with no models installed): show that rather than
# hardcoded suggestions the crew can't actually run. The picker handles an
# empty list by prompting for manual entry.
vendor = _from_vendor(provider_key)
if vendor is not None:
result = _finalize(vendor, label_map)
if result:
_write_catalog_cache(provider_key, result, source="dynamic")
return result
# Vendor tier unavailable. The LiteLLM feed lags real releases, so only
# reach for it when we have no curated fallback — never override the fallback.
entries = _from_litellm(provider_key) if not fallback else None
result = _finalize(entries, label_map) if entries else []
if result:
_write_catalog_cache(provider_key, result, source="dynamic")
return result
# Nothing fresher than the curated list. Cache it briefly (negative cache)
# so a failed vendor/LiteLLM fetch isn't retried on every subsequent call.
# Skip Ollama: it's a local, fast-failing server, so re-probing is cheap and
# avoids serving suggestions after the server comes up within the TTL.
if fallback and provider_key != "ollama":
_write_catalog_cache(provider_key, fallback, source="fallback")
return fallback
# ── Tier 1: vendor APIs ──────────────────────────────────────────
def _from_vendor(provider_key: str) -> list[dict[str, Any]] | None:
"""Fetch models from the vendor.
Returns the model list on a successful fetch — **including an empty list**,
which is meaningful (e.g. a reachable Ollama server with nothing installed).
Returns ``None`` only when the vendor tier is unavailable: no fetcher, no
API key, or the request failed.
"""
fetcher = _VENDOR_FETCHERS.get(provider_key)
if fetcher is None:
return None
api_key = _provider_api_key(provider_key)
if _PROVIDER_KEY_ENV.get(provider_key) and not api_key:
# Provider needs a key and none is set — skip to the next tier.
return None
try:
return fetcher(api_key)
except Exception:
# Network error, auth failure, unexpected payload — degrade quietly.
return None
def _fetch_openai(api_key: str | None) -> list[dict[str, Any]]:
return _fetch_openai_compatible("https://api.openai.com/v1", api_key)
def _fetch_groq(api_key: str | None) -> list[dict[str, Any]]:
return _fetch_openai_compatible("https://api.groq.com/openai/v1", api_key)
def _fetch_cerebras(api_key: str | None) -> list[dict[str, Any]]:
return _fetch_openai_compatible("https://api.cerebras.ai/v1", api_key)
def _fetch_openai_compatible(
base_url: str, api_key: str | None
) -> list[dict[str, Any]]:
"""Parse an OpenAI-shaped ``GET /models`` response."""
data = _http_get_json(
f"{base_url}/models",
headers={"Authorization": f"Bearer {api_key}"},
)
entries: list[dict[str, Any]] = []
for item in data.get("data", []):
model_id = item.get("id")
if not model_id or not _is_chat_model(model_id) or _is_fine_tune(model_id):
continue
created = _as_float(item.get("created"))
entries.append(_entry(model_id, _humanize(model_id), created=created))
return entries
def _fetch_anthropic(api_key: str | None) -> list[dict[str, Any]]:
data = _http_get_json(
"https://api.anthropic.com/v1/models",
headers={"x-api-key": api_key or "", "anthropic-version": "2023-06-01"},
)
entries: list[dict[str, Any]] = []
for item in data.get("data", []):
model_id = item.get("id")
if not model_id:
continue
label = item.get("display_name") or _humanize(model_id)
created = _parse_iso(item.get("created_at"))
entries.append(_entry(model_id, label, created=created))
return entries
def _fetch_gemini(api_key: str | None) -> list[dict[str, Any]]:
entries: list[dict[str, Any]] = []
params: dict[str, Any] = {"key": api_key or "", "pageSize": 200}
# models.list is paginated and not guaranteed newest-first, so walk pages
# (bounded) to see the full set — _finalize does the sort + truncation.
for _ in range(10):
try:
data = _http_get_json(
"https://generativelanguage.googleapis.com/v1beta/models",
params=params,
)
except Exception:
# Later-page failure: keep the models already gathered. First-page
# failure (nothing gathered yet) is a real outage — re-raise so the
# caller falls back to the curated list rather than mistaking it for
# a successful empty result.
if entries:
break
raise
for item in data.get("models", []):
methods = item.get("supportedGenerationMethods") or []
if "generateContent" not in methods:
continue
name = (item.get("name") or "").removeprefix("models/")
if not name or not _is_chat_model(name) or "aqa" in name:
continue
label = item.get("displayName") or _humanize(name)
# Gemini has no timestamp; rank by the version in name/version.
version_hint = f"{name} {item.get('version') or ''}"
entries.append(_entry(name, label, version_hint=version_hint))
token = data.get("nextPageToken")
if not token:
break
params = {"key": api_key or "", "pageSize": 200, "pageToken": token}
return entries
def _ollama_base() -> str:
"""Resolve the Ollama server base URL from the environment.
Checks ``OLLAMA_API_BASE`` / ``API_BASE`` (what LiteLLM and the generated
crew use) first, then ``OLLAMA_HOST`` (the Ollama runtime convention), so a
user who only set ``OLLAMA_HOST`` sees models from the right server.
"""
base = (
os.environ.get("OLLAMA_API_BASE")
or os.environ.get("API_BASE")
or os.environ.get("OLLAMA_HOST")
or "http://localhost:11434"
).strip()
# OLLAMA_HOST is often scheme-less (e.g. "127.0.0.1:11434").
if "://" not in base:
base = f"http://{base}"
return base.rstrip("/")
def _fetch_ollama(_api_key: str | None) -> list[dict[str, Any]]:
"""List models installed on the local Ollama server (no API key)."""
data = _http_get_json(f"{_ollama_base()}/api/tags")
entries: list[dict[str, Any]] = []
for item in data.get("models", []):
model_id = item.get("model") or item.get("name")
if not model_id or not _is_chat_model(model_id) or _is_fine_tune(model_id):
# /api/tags lists everything installed, including embedding models.
continue
# Ollama returns an ISO 8601 modified_at we can rank by.
created = _parse_iso(item.get("modified_at"))
entries.append(_entry(model_id, _humanize(model_id), created=created))
return entries
_VENDOR_FETCHERS: dict[str, Callable[[str | None], list[dict[str, Any]]]] = {
"openai": _fetch_openai,
"anthropic": _fetch_anthropic,
"gemini": _fetch_gemini,
"groq": _fetch_groq,
"cerebras": _fetch_cerebras,
"ollama": _fetch_ollama,
}
# ── Tier 2: LiteLLM feed ─────────────────────────────────────────
# Process-level memo so a single CLI run attempts the LiteLLM download at most
# once — repeated picker calls otherwise each incur a multi-second timeout when
# the feed is stale/unreachable. Reset via _reset_litellm_memo() in tests.
_UNSET: Any = object()
_litellm_memo: Any = _UNSET
def _reset_litellm_memo() -> None:
"""Clear the process-level LiteLLM memo (test hook)."""
global _litellm_memo
_litellm_memo = _UNSET
def _from_litellm(provider_key: str) -> list[dict[str, Any]] | None:
"""Build chat-model entries for ``provider_key`` from the LiteLLM feed."""
data = _load_litellm_data()
# A corrupt feed (non-mapping JSON root) must not crash the picker.
if not isinstance(data, dict):
return None
entries: list[dict[str, Any]] = []
for model_name, props in data.items():
if not isinstance(props, dict):
continue
# `litellm_provider` can be present-but-null in the feed; coerce before
# string ops so a null value is skipped rather than raising.
if (props.get("litellm_provider") or "").strip().lower() != provider_key:
continue
if props.get("mode") != "chat":
continue
# LiteLLM keys are sometimes prefixed with the provider; the picker
# re-adds ``provider/`` itself, so strip a leading one to avoid dupes.
model_id = model_name
if model_id.startswith(f"{provider_key}/"):
model_id = model_id[len(provider_key) + 1 :]
if not model_id:
continue
entries.append(_entry(model_id, _humanize(model_id), version_hint=model_id))
return entries or None
def _load_litellm_data() -> dict[str, Any] | None:
"""Return the LiteLLM feed, memoized once per process (see _litellm_memo)."""
global _litellm_memo
if _litellm_memo is _UNSET:
_litellm_memo = _fetch_litellm_data()
memoized: dict[str, Any] | None = _litellm_memo
return memoized
def _fetch_litellm_data() -> dict[str, Any] | None:
"""Read the cached LiteLLM feed, fetching it once if the cache is cold."""
cache_file = _litellm_cache_file()
fresh = (
cache_file.exists()
and (time.time() - cache_file.stat().st_mtime) < _LITELLM_TTL
)
if fresh:
data = _read_json(cache_file)
# A corrupt/non-mapping fresh cache must not block a recoverable
# download — only short-circuit on a usable mapping.
if isinstance(data, dict) and data:
return data
try:
data = _http_get_json(JSON_URL)
except Exception:
# Fall back to a stale cache if we have one, else give up on this tier.
return _read_json(cache_file)
# Best-effort cache write; a failure (e.g. read-only home) is non-fatal
# since we already hold the freshly-fetched data.
with contextlib.suppress(OSError):
cache_file.parent.mkdir(parents=True, exist_ok=True)
cache_file.write_text(json.dumps(data), encoding="utf-8")
return data
# ── Ranking + labelling ──────────────────────────────────────────
def _finalize(
entries: list[dict[str, Any]], label_map: dict[str, str]
) -> list[tuple[str, str]]:
"""Sort newest-first, dedupe, relabel with curated names, and truncate."""
entries.sort(key=lambda e: e["sort"], reverse=True)
seen: set[str] = set()
out: list[tuple[str, str]] = []
for entry in entries:
model_id = entry["id"]
if model_id in seen:
continue
seen.add(model_id)
label = label_map.get(model_id) or entry["label"]
out.append((model_id, label))
if len(out) >= MAX_MODELS:
break
return out
def _entry(
model_id: str,
label: str,
*,
created: float = 0.0,
version_hint: str | None = None,
) -> dict[str, Any]:
"""Build a rankable catalog entry.
``sort`` is a comparable tuple ``(created, date_int, version_tuple)`` so a
real vendor timestamp wins, then a date embedded in the id, then the numeric
version. Types line up positionally, so entries compare cleanly.
"""
date_int, version = _version_key(version_hint or model_id)
return {
"id": model_id,
"label": label,
"sort": (created, date_int, version),
}
_DATE_RE = re.compile(r"(20\d{2})[-_]?(0[1-9]|1[0-2])[-_]?(0[1-9]|[12]\d|3[01])")
_NUM_RE = re.compile(r"\d+")
def _version_key(text: str) -> tuple[int, tuple[int, ...]]:
"""Extract a ``(date_int, version_tuple)`` sort key from a model id.
A trailing/embedded ``YYYYMMDD`` (or ``YYYY-MM-DD``) becomes ``date_int``;
remaining numbers become the version tuple. ``claude-opus-4-6`` → version
``(4, 6)``; ``claude-3-5-sonnet-20241022`` → date ``20241022`` version
``(3, 5)``.
"""
text = text or ""
date_int = 0
match = _DATE_RE.search(text)
if match:
date_int = int(match.group(1) + match.group(2) + match.group(3))
text = _DATE_RE.sub(" ", text)
version = tuple(int(n) for n in _NUM_RE.findall(text)[:4])
return date_int, version
def _is_chat_model(model_id: str) -> bool:
"""Heuristically reject embedding/audio/image/etc. models by their id."""
lowered = model_id.lower()
return not any(marker in lowered for marker in _NON_CHAT_MARKERS)
def _is_fine_tune(model_id: str) -> bool:
"""A user fine-tune or training checkpoint (``ft:...`` / ``...:ckpt-step-N``).
These are account-specific artifacts: they clutter the picker, crowd out the
foundation models (their recent ``created`` timestamps rank them first), and
humanize into unreadable labels. Excluded from the auto-list; a user who
wants one can still enter it via the picker's "Other" option.
"""
lowered = model_id.lower()
return lowered.startswith("ft:") or ":ckpt" in lowered
_SIZE_RE = re.compile(r"^\d+(?:\.\d+)?[bmk]$") # 8b, 70b, 1.5b, 120m, 32k
_OSERIES_RE = re.compile(r"^o\d+$") # o1, o3, o4 — kept lowercase (OpenAI brand)
def _humanize(model_id: str) -> str:
"""Derive a readable label from a raw model id.
Best-effort only — vendor display names and the curated label map take
precedence. Drops embedded dates and applies light casing so raw ids read
cleanly: ``gpt-oss-120b`` → ``GPT OSS 120B``, ``qwen3-32b`` → ``Qwen3 32B``,
``deepseek-r1:671b`` → ``DeepSeek R1 671B``, ``o3-mini`` → ``o3 Mini``.
"""
base = model_id.split("/")[-1]
# Drop embedded release dates — they're noise in a label, and the picker
# already shows the full model id alongside it.
base = _DATE_RE.sub(" ", base)
words: list[str] = []
# Split on separators including ``:`` so Ollama tags (llama3.3:70b) read well.
for part in re.split(r"[-_\s:]+", base):
if not part:
continue
low = part.lower()
if low in _ACRONYMS:
words.append(_ACRONYMS[low])
elif low in _BRAND_TOKENS:
words.append(_BRAND_TOKENS[low])
elif _SIZE_RE.match(low):
words.append(low[:-1] + low[-1].upper()) # 70b -> 70B
elif _OSERIES_RE.match(low):
words.append(low) # o3 stays lowercase
elif part[0].isalpha():
# Capitalize the leading letter, preserve the rest (so a fused
# family+version keeps its digits): qwen3 -> Qwen3, mini -> Mini.
words.append(part[0].upper() + part[1:])
else:
words.append(part) # starts with a digit (4o, 4.1, 0905) — leave as-is
return " ".join(words) or base
# ── HTTP + parsing helpers ───────────────────────────────────────
def _http_get_json(
url: str,
*,
headers: dict[str, str] | None = None,
params: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""GET ``url`` and return parsed JSON, with a short timeout and TLS verify."""
ssl_config = os.environ.get("SSL_CERT_FILE") or certifi.where()
response = httpx.get(
url,
headers=headers,
params=params,
timeout=_TIMEOUT,
verify=ssl_config,
follow_redirects=True,
)
response.raise_for_status()
result: dict[str, Any] = response.json()
return result
def _parse_iso(value: Any) -> float:
"""Parse an ISO 8601 timestamp to an epoch float; ``0.0`` on failure."""
if not value or not isinstance(value, str):
return 0.0
from datetime import datetime
try:
return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
except ValueError:
return 0.0
def _as_float(value: Any) -> float:
try:
return float(value)
except (TypeError, ValueError):
return 0.0
def _read_json(path: Path) -> dict[str, Any] | None:
try:
data: dict[str, Any] = json.loads(path.read_text(encoding="utf-8"))
return data
except (OSError, json.JSONDecodeError):
return None
# ── Caching ──────────────────────────────────────────────────────
def _cache_dir() -> Path:
return Path.home() / ".crewai"
def _catalog_cache_file() -> Path:
return _cache_dir() / "model_catalog_cache.json"
def _litellm_cache_file() -> Path:
# Shared with crewai_cli.provider so both flows warm the same cache.
return _cache_dir() / "provider_cache.json"
def _cache_key(provider_key: str) -> str:
"""Cache key for a provider's resolved model list.
Includes the inputs that change what a fetch would return, so a cached
entry is only reused when those inputs still match:
- Ollama lists models from a base URL that can change between runs.
- Whether the vendor's API key is present flips between a live fetch and
the negatively-cached fallback — so a key added after a no-key call is
not shadowed by the cached fallback.
"""
if provider_key == "ollama":
return f"ollama@{_ollama_base()}"
suffix = "key" if _provider_api_key(provider_key) else "nokey"
return f"{provider_key}#{suffix}"
def _read_catalog_cache(provider_key: str) -> list[tuple[str, str]] | None:
"""Return a fresh cached catalog for ``provider_key``, or ``None``."""
payload = _read_json(_catalog_cache_file())
if not isinstance(payload, dict):
return None
entry = payload.get(_cache_key(provider_key))
if not isinstance(entry, dict):
return None
# Fallback (negative) entries expire fast; dynamic ones live the full TTL.
ttl = _NEGATIVE_TTL if entry.get("source") == "fallback" else _CATALOG_TTL
if (time.time() - _as_float(entry.get("ts"))) >= ttl:
return None
models = entry.get("models")
if not isinstance(models, list) or not models:
return None
try:
return [(str(m[0]), str(m[1])) for m in models]
except (IndexError, TypeError):
return None
def _write_catalog_cache(
provider_key: str, models: list[tuple[str, str]], *, source: str
) -> None:
cache_file = _catalog_cache_file()
payload = _read_json(cache_file)
if not isinstance(payload, dict):
payload = {}
payload[_cache_key(provider_key)] = {
"ts": time.time(),
"source": source,
"models": [[model_id, label] for model_id, label in models],
}
# Best-effort cache write; a failure (e.g. read-only home) is non-fatal.
with contextlib.suppress(OSError):
cache_file.parent.mkdir(parents=True, exist_ok=True)
cache_file.write_text(json.dumps(payload), encoding="utf-8")

551
lib/cli/tests/test_model_catalog.py Normal file
View File

@@ -0,0 +1,551 @@
"""Tests for the dynamic model catalog used by the crew-creation wizard."""
from __future__ import annotations
import json
import pytest
import crewai_cli.model_catalog as mc
_ALL_KEY_ENVS = [
"OPENAI_API_KEY",
"ANTHROPIC_API_KEY",
"GEMINI_API_KEY",
"GOOGLE_API_KEY",
"GROQ_API_KEY",
"CEREBRAS_API_KEY",
"OLLAMA_API_BASE",
"API_BASE",
"OLLAMA_HOST",
]
FALLBACK_ANTHROPIC = [
("claude-opus-4-6", "Claude Opus 4.6"),
("claude-sonnet-4-6", "Claude Sonnet 4.6"),
]
@pytest.fixture(autouse=True)
def isolated_env(monkeypatch, tmp_path):
"""Point the cache at a temp dir and clear provider keys for every test."""
monkeypatch.setattr(mc, "_cache_dir", lambda: tmp_path)
mc._reset_litellm_memo() # clear the process-level LiteLLM memo per test
for key in _ALL_KEY_ENVS:
monkeypatch.delenv(key, raising=False)
# ── version / label helpers ──────────────────────────────────────
def test_version_key_parses_embedded_date():
date_int, version = mc._version_key("claude-3-5-sonnet-20241022")
assert date_int == 20241022
assert version == (3, 5)
def test_version_key_parses_dashed_date():
date_int, _ = mc._version_key("gpt-4o-2024-08-06")
assert date_int == 20240806
def test_version_key_version_only():
date_int, version = mc._version_key("claude-opus-4-6")
assert date_int == 0
assert version == (4, 6)
def test_version_key_ranks_newer_higher():
older = mc._version_key("claude-sonnet-4-5")
newer = mc._version_key("claude-sonnet-4-6")
assert newer > older
def test_is_chat_model_rejects_non_chat():
assert mc._is_chat_model("gpt-4.1-mini")
assert not mc._is_chat_model("text-embedding-3-large")
assert not mc._is_chat_model("whisper-1")
assert not mc._is_chat_model("dall-e-3")
def test_search_substring_not_treated_as_non_chat():
# 'search' must not drop legitimate completion models: a token like
# *-search-preview, or 'research' (which contains 'search' as a substring).
assert mc._is_chat_model("gpt-4o-search-preview")
assert mc._is_chat_model("o3-deep-research")
# genuine non-chat markers still filter
assert not mc._is_chat_model("text-embedding-3-large")
def test_humanize():
assert mc._humanize("gpt-4.1-mini") == "GPT 4.1 Mini"
assert mc._humanize("anthropic/claude-opus-4-6") == "Claude Opus 4 6"
# size suffixes uppercased, acronyms/brands cased, o-series preserved, ':' split
assert mc._humanize("openai/gpt-oss-120b") == "GPT OSS 120B"
assert mc._humanize("qwen/qwen3-32b") == "Qwen3 32B"
assert mc._humanize("deepseek-r1-distill-llama-70b") == "DeepSeek R1 Distill Llama 70B"
assert mc._humanize("o3-mini") == "o3 Mini"
assert mc._humanize("chatgpt-4o-latest") == "ChatGPT 4o Latest"
assert mc._humanize("llama3.3:70b") == "Llama3.3 70B"
assert mc._humanize("gemma2-9b-it") == "Gemma2 9B IT"
# ── vendor tier ──────────────────────────────────────────────────
def test_vendor_anthropic_ranks_by_date_and_uses_display_name(monkeypatch):
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test")
payload = {
"data": [
{
"id": "claude-3-5-sonnet-20240620",
"display_name": "Claude 3.5 Sonnet (old)",
"created_at": "2024-06-20T00:00:00Z",
},
{
"id": "claude-opus-4-6",
"display_name": "Claude Opus 4.6",
"created_at": "2026-02-01T00:00:00Z",
},
{
"id": "claude-haiku-4-5-20251001",
"display_name": "Claude Haiku 4.5",
"created_at": "2025-10-01T00:00:00Z",
},
]
}
monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: payload)
models = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
# Newest first by created_at, display names preserved.
assert models[0] == ("claude-opus-4-6", "Claude Opus 4.6")
assert models[1] == ("claude-haiku-4-5-20251001", "Claude Haiku 4.5")
assert models[2] == ("claude-3-5-sonnet-20240620", "Claude 3.5 Sonnet (old)")
def test_vendor_openai_filters_non_chat_models(monkeypatch):
monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
payload = {
"data": [
{"id": "gpt-4.1", "created": 1_700_000_000},
{"id": "text-embedding-3-large", "created": 1_800_000_000},
{"id": "whisper-1", "created": 1_800_000_000},
{"id": "gpt-5.5", "created": 1_750_000_000},
]
}
monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: payload)
models = mc.get_provider_models("openai", [])
ids = [m for m, _ in models]
assert ids == ["gpt-5.5", "gpt-4.1"] # embeddings/whisper dropped, newest first
def test_vendor_gemini_requires_generate_content(monkeypatch):
monkeypatch.setenv("GEMINI_API_KEY", "key")
payload = {
"models": [
{
"name": "models/gemini-2.5-pro",
"displayName": "Gemini 2.5 Pro",
"supportedGenerationMethods": ["generateContent"],
},
{
"name": "models/text-embedding-004",
"displayName": "Embedding",
"supportedGenerationMethods": ["embedContent"],
},
{
"name": "models/gemini-1.5-pro",
"displayName": "Gemini 1.5 Pro",
"supportedGenerationMethods": ["generateContent"],
},
]
}
monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: payload)
models = mc.get_provider_models("gemini", [])
ids = [m for m, _ in models]
# "models/" prefix stripped, embedding excluded, newer version first.
assert ids == ["gemini-2.5-pro", "gemini-1.5-pro"]
def test_openai_excludes_fine_tunes_and_checkpoints(monkeypatch):
# Fine-tunes/checkpoints have recent `created` timestamps and would otherwise
# crowd out (and rank above) the base models — they must be excluded so the
# picker shows clean foundation models.
monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
payload = {
"data": [
{"id": "ft:gpt-4o-mini-2024-07-18:crewai::DyJG86uF", "created": 1_900_000_000},
{
"id": "ft:gpt-4o-mini-2024-07-18:crewai::DyJG7Q9N:ckpt-step-84",
"created": 1_900_000_001,
},
{"id": "gpt-5.5", "created": 1_800_000_000},
{"id": "gpt-4.1", "created": 1_700_000_000},
]
}
monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: payload)
ids = [m for m, _ in mc.get_provider_models("openai", [])]
assert ids == ["gpt-5.5", "gpt-4.1"] # fine-tunes + checkpoints dropped
def test_vendor_gemini_paginates(monkeypatch):
monkeypatch.setenv("GEMINI_API_KEY", "key")
pages = {
None: {
"models": [
{
"name": "models/gemini-3.5-flash",
"displayName": "Gemini 3.5 Flash",
"supportedGenerationMethods": ["generateContent"],
}
],
"nextPageToken": "p2",
},
"p2": {
"models": [
{
"name": "models/gemini-2.5-pro",
"displayName": "Gemini 2.5 Pro",
"supportedGenerationMethods": ["generateContent"],
}
]
},
}
def fetch(url, headers=None, params=None):
return pages[(params or {}).get("pageToken")]
monkeypatch.setattr(mc, "_http_get_json", fetch)
ids = sorted(m for m, _ in mc.get_provider_models("gemini", []))
# Both pages contributed (newest-first ranking is _finalize's job).
assert ids == ["gemini-2.5-pro", "gemini-3.5-flash"]
def test_vendor_gemini_first_page_error_uses_fallback(monkeypatch):
# A total (first-page) Gemini failure with a key set must fall back to the
# curated list, not be mistaken for a successful empty result.
monkeypatch.setenv("GEMINI_API_KEY", "key")
def boom(*a, **k):
raise RuntimeError("gemini down")
monkeypatch.setattr(mc, "_http_get_json", boom)
models = mc.get_provider_models("gemini", [("gemini-x", "Gemini X")])
assert models == [("gemini-x", "Gemini X")]
def test_vendor_gemini_keeps_partial_on_later_page_error(monkeypatch):
monkeypatch.setenv("GEMINI_API_KEY", "key")
def fetch(url, headers=None, params=None):
if (params or {}).get("pageToken"):
raise RuntimeError("page 2 down")
return {
"models": [
{
"name": "models/gemini-3.5-flash",
"displayName": "Gemini 3.5 Flash",
"supportedGenerationMethods": ["generateContent"],
}
],
"nextPageToken": "p2",
}
monkeypatch.setattr(mc, "_http_get_json", fetch)
# Page-1 models are kept; the later-page error doesn't force the fallback.
models = mc.get_provider_models("gemini", [("fallback-x", "Fallback X")])
assert [m for m, _ in models] == ["gemini-3.5-flash"]
def test_ollama_empty_response_not_filled_with_fallback(monkeypatch):
# A reachable Ollama with nothing installed -> empty (manual entry), not the
# curated suggestions the crew can't actually run.
monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: {"models": []})
assert mc.get_provider_models("ollama", [("llama3.3", "Llama 3.3")]) == []
def test_ollama_unreachable_uses_fallback(monkeypatch):
# Server down (fetch raises) is different from empty -> fall back to suggestions.
def boom(*a, **k):
raise RuntimeError("connection refused")
monkeypatch.setattr(mc, "_http_get_json", boom)
models = mc.get_provider_models("ollama", [("llama3.3", "Llama 3.3")])
assert models == [("llama3.3", "Llama 3.3")]
def test_ollama_excludes_embedding_models(monkeypatch):
# /api/tags lists everything installed, including embeddings — filter them.
monkeypatch.setattr(
mc,
"_http_get_json",
lambda *a, **k: {
"models": [
{"model": "llama3.3:70b"},
{"model": "nomic-embed-text"},
{"model": "mxbai-embed-large"},
]
},
)
ids = [m for m, _ in mc.get_provider_models("ollama", [])]
assert ids == ["llama3.3:70b"]
def test_ollama_base_honors_ollama_host(monkeypatch):
# OLLAMA_HOST (scheme-less runtime convention) is resolved with a scheme.
monkeypatch.setenv("OLLAMA_HOST", "10.0.0.5:11434")
assert mc._ollama_base() == "http://10.0.0.5:11434"
def test_ollama_recovery_not_blocked_by_negative_cache(monkeypatch):
# Ollama down -> fallback, but not negatively cached; once the server is up
# the next call fetches live models rather than serving suggestions.
calls = {"n": 0}
def flaky(*a, **k):
calls["n"] += 1
if calls["n"] == 1:
raise RuntimeError("connection refused")
return {"models": [{"model": "llama-installed"}]}
monkeypatch.setattr(mc, "_http_get_json", flaky)
first = mc.get_provider_models("ollama", [("llama3.3", "Llama 3.3")])
assert first == [("llama3.3", "Llama 3.3")] # down -> fallback (not cached)
second = mc.get_provider_models("ollama", [("llama3.3", "Llama 3.3")])
assert [m for m, _ in second] == ["llama-installed"] # recovered live
def test_gemini_honors_google_api_key(monkeypatch):
# GOOGLE_API_KEY (equivalent to GEMINI_API_KEY in crewai) enables the live tier.
monkeypatch.setenv("GOOGLE_API_KEY", "key")
monkeypatch.setattr(
mc,
"_http_get_json",
lambda *a, **k: {
"models": [
{
"name": "models/gemini-3.5-flash",
"displayName": "Gemini 3.5 Flash",
"supportedGenerationMethods": ["generateContent"],
}
]
},
)
models = mc.get_provider_models("gemini", [("gemini-x", "Gemini X")])
assert [m for m, _ in models] == ["gemini-3.5-flash"] # live, not fallback
def test_curated_label_overrides_raw_vendor_label(monkeypatch):
monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
payload = {"data": [{"id": "gpt-5.5", "created": 1}]}
monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: payload)
models = mc.get_provider_models("openai", [("gpt-5.5", "GPT-5.5 (curated)")])
assert models == [("gpt-5.5", "GPT-5.5 (curated)")]
def test_truncates_to_max_models(monkeypatch):
monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
payload = {
"data": [{"id": f"gpt-test-{i}", "created": i} for i in range(20)]
}
monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: payload)
models = mc.get_provider_models("openai", [])
assert len(models) == mc.MAX_MODELS
# ── litellm tier ─────────────────────────────────────────────────
def test_litellm_tier_for_uncurated_provider(monkeypatch):
# A provider with no curated fallback ([]) -> the LiteLLM feed is consulted.
litellm_data = {
"claude-opus-4-6": {"litellm_provider": "anthropic", "mode": "chat"},
"claude-sonnet-4-5": {"litellm_provider": "anthropic", "mode": "chat"},
"voyage-embed": {"litellm_provider": "anthropic", "mode": "embedding"},
"gpt-4.1": {"litellm_provider": "openai", "mode": "chat"},
}
mc._litellm_cache_file().write_text(json.dumps(litellm_data), encoding="utf-8")
models = mc.get_provider_models("anthropic", []) # empty == uncurated
ids = [m for m, _ in models]
# Only anthropic chat models, embedding + other providers excluded.
assert ids == ["claude-opus-4-6", "claude-sonnet-4-5"]
def test_null_litellm_provider_does_not_crash(monkeypatch):
# A present-but-null litellm_provider must be skipped, not raise.
litellm_data = {
"weird-model": {"litellm_provider": None, "mode": "chat"},
"anthropic.claude-v2": {"litellm_provider": "bedrock", "mode": "chat"},
}
mc._litellm_cache_file().write_text(json.dumps(litellm_data), encoding="utf-8")
models = mc.get_provider_models("bedrock", [])
assert [m for m, _ in models] == ["anthropic.claude-v2"]
def test_litellm_strips_provider_prefix(monkeypatch):
litellm_data = {
"gemini/gemini-1.5-pro": {"litellm_provider": "gemini", "mode": "chat"},
}
mc._litellm_cache_file().write_text(json.dumps(litellm_data), encoding="utf-8")
models = mc.get_provider_models("gemini", [])
assert models == [("gemini-1.5-pro", "Gemini 1.5 Pro")]
# ── fallback + caching ───────────────────────────────────────────
def test_falls_back_when_everything_fails(monkeypatch):
# No key, no litellm cache, network raises -> curated fallback verbatim.
def boom(*a, **k):
raise RuntimeError("network down")
monkeypatch.setattr(mc, "_http_get_json", boom)
models = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
assert models == FALLBACK_ANTHROPIC
def test_result_is_cached(monkeypatch):
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test")
calls = {"n": 0}
def fetch(*a, **k):
calls["n"] += 1
return {"data": [{"id": "claude-opus-4-6", "created_at": "2026-01-01T00:00:00Z"}]}
monkeypatch.setattr(mc, "_http_get_json", fetch)
first = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
# Second call must hit the cache and not touch the network again.
monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: pytest.fail("refetched"))
second = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
assert first == second
assert calls["n"] == 1
def test_curated_fallback_preferred_over_litellm(monkeypatch):
# The feed lags real releases, so a non-empty curated fallback must win even
# when a fresh LiteLLM cache is present (regression: Anthropic's feed lacked
# Fable 5 / Opus 4.8 / Sonnet 5).
monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: pytest.fail("no net"))
litellm_data = {
"claude-opus-4-6": {"litellm_provider": "anthropic", "mode": "chat"},
}
mc._litellm_cache_file().write_text(json.dumps(litellm_data), encoding="utf-8")
models = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
assert models == FALLBACK_ANTHROPIC
def test_added_key_bypasses_negative_cache(monkeypatch):
# A no-key call negatively-caches the fallback; adding a key afterwards must
# fetch live models rather than serve the cached fallback (distinct cache key).
first = mc.get_provider_models("openai", [("gpt-x", "GPT X")])
assert first == [("gpt-x", "GPT X")] # no key -> fallback
monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
monkeypatch.setattr(
mc, "_http_get_json", lambda *a, **k: {"data": [{"id": "gpt-5.5", "created": 1}]}
)
second = mc.get_provider_models("openai", [("gpt-x", "GPT X")])
assert [m for m, _ in second] == ["gpt-5.5"] # live fetch, not cached fallback
def test_invalid_litellm_cache_falls_through_to_download(monkeypatch):
# A corrupt-but-fresh cache must neither crash the picker nor block a
# recoverable download — it falls through and refetches.
mc._litellm_cache_file().write_text("[1, 2, 3]", encoding="utf-8")
monkeypatch.setattr(
mc,
"_http_get_json",
lambda *a, **k: {
"anthropic.claude-v2": {"litellm_provider": "bedrock", "mode": "chat"}
},
)
models = mc.get_provider_models("bedrock", [])
assert [m for m, _ in models] == ["anthropic.claude-v2"] # recovered via download
def test_litellm_fetch_attempted_once_per_process(monkeypatch):
# With no cache and a failing download, the feed is fetched at most once per
# process — repeated lookups (across providers) must not re-hit the network.
calls = {"n": 0}
def boom(*a, **k):
calls["n"] += 1
raise RuntimeError("offline")
monkeypatch.setattr(mc, "_http_get_json", boom)
mc.get_provider_models("bedrock", [])
mc.get_provider_models("azure", [])
assert calls["n"] == 1 # memoized after the first failed attempt
def test_litellm_fills_uncurated_bedrock(monkeypatch):
# No vendor fetcher and no curated fallback -> LiteLLM feed fills the gap.
monkeypatch.setattr(mc, "_http_get_json", lambda *a, **k: pytest.fail("no net"))
litellm_data = {
"anthropic.claude-v2": {"litellm_provider": "bedrock", "mode": "chat"},
}
mc._litellm_cache_file().write_text(json.dumps(litellm_data), encoding="utf-8")
models = mc.get_provider_models("bedrock", [])
assert models == [("anthropic.claude-v2", "Anthropic.claude V2")]
def test_failed_fetch_is_negatively_cached(monkeypatch):
# A failed vendor fetch must not be retried on every call — the fallback is
# cached briefly so the picker doesn't re-hit the timeout-prone endpoint.
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test")
calls = {"n": 0}
def boom(*a, **k):
calls["n"] += 1
raise RuntimeError("down")
monkeypatch.setattr(mc, "_http_get_json", boom)
first = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
second = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
assert first == second == FALLBACK_ANTHROPIC
assert calls["n"] == 1 # second call served from the negative cache
def test_bad_cache_json_does_not_crash(monkeypatch):
# A corrupt cache whose root is not a mapping must not raise (get_provider_models
# is documented to never raise).
mc._catalog_cache_file().write_text("[1, 2, 3]", encoding="utf-8")
models = mc.get_provider_models("anthropic", FALLBACK_ANTHROPIC)
assert models == FALLBACK_ANTHROPIC
def test_ollama_cache_keyed_by_base(monkeypatch):
# Changing OLLAMA_API_BASE must not serve the previous host's cached models.
monkeypatch.setenv("OLLAMA_API_BASE", "http://host-a:11434")
monkeypatch.setattr(
mc, "_http_get_json", lambda *a, **k: {"models": [{"model": "llama-a"}]}
)
first = mc.get_provider_models("ollama", [])
assert [m for m, _ in first] == ["llama-a"]
monkeypatch.setenv("OLLAMA_API_BASE", "http://host-b:11434")
monkeypatch.setattr(
mc, "_http_get_json", lambda *a, **k: {"models": [{"model": "llama-b"}]}
)
second = mc.get_provider_models("ollama", [])
assert [m for m, _ in second] == ["llama-b"] # not the host-a cache

2
lib/crewai-core/src/crewai_core/telemetry.py
View File

@@ -264,10 +264,12 @@ class Telemetry:
def flow_creation_span(self, flow_name: str) -> None:
"""Records the creation of a new flow."""
from crewai_core.version import get_crewai_version
def _operation() -> None:
tracer = trace.get_tracer("crewai.telemetry")
span = tracer.start_span("Flow Creation")
self._add_attribute(span, "crewai_version", get_crewai_version())
self._add_attribute(span, "flow_name", flow_name)
close_span(span)

28
lib/crewai-core/tests/test_smoke.py
View File

@@ -233,3 +233,31 @@ def test_core_telemetry_records_feature_usage(
tracer.start_span.assert_called_once_with("Feature Usage")
span.set_attribute.assert_any_call("feature", "cli_usage:view_traces")
span.end.assert_called_once()
def test_core_telemetry_records_flow_creation_version(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from crewai_core.telemetry import Telemetry
Telemetry._instance = None
monkeypatch.delenv("OTEL_SDK_DISABLED", raising=False)
monkeypatch.delenv("CREWAI_DISABLE_TELEMETRY", raising=False)
monkeypatch.delenv("CREWAI_DISABLE_TRACKING", raising=False)
monkeypatch.setattr("crewai_core.version.get_crewai_version", lambda: "1.0.0")
tracer = Mock()
span = Mock()
tracer.start_span.return_value = span
monkeypatch.setattr(
"crewai_core.telemetry.trace.get_tracer",
lambda _name: tracer,
)
telemetry = Telemetry()
telemetry.flow_creation_span("ResearchFlow")
tracer.start_span.assert_called_once_with("Flow Creation")
span.set_attribute.assert_any_call("crewai_version", "1.0.0")
span.set_attribute.assert_any_call("flow_name", "ResearchFlow")
span.end.assert_called_once()

6
lib/crewai-tools/src/crewai_tools/tools/tavily_extractor_tool/tavily_extractor_tool.py
View File

@@ -87,9 +87,11 @@ class TavilyExtractorTool(BaseTool):
"""
super().__init__(**kwargs)
if TAVILY_AVAILABLE:
self.client = TavilyClient(api_key=self.api_key, proxies=self.proxies)
self.client = TavilyClient(
api_key=self.api_key, proxies=self.proxies, client_name="crewai"
)
self.async_client = AsyncTavilyClient(
api_key=self.api_key, proxies=self.proxies
api_key=self.api_key, proxies=self.proxies, client_name="crewai"
)
else:
try:

6
lib/crewai-tools/src/crewai_tools/tools/tavily_get_research_tool/tavily_get_research_tool.py
View File

@@ -54,8 +54,10 @@ class TavilyGetResearchTool(BaseTool):
super().__init__(**kwargs)
if TAVILY_AVAILABLE:
api_key = os.getenv("TAVILY_API_KEY")
self._client = TavilyClient(api_key=api_key)
self._async_client = AsyncTavilyClient(api_key=api_key)
self._client = TavilyClient(api_key=api_key, client_name="crewai")
self._async_client = AsyncTavilyClient(
api_key=api_key, client_name="crewai"
)
else:
try:
import subprocess

6
lib/crewai-tools/src/crewai_tools/tools/tavily_research_tool/tavily_research_tool.py
View File

@@ -90,8 +90,10 @@ class TavilyResearchTool(BaseTool):
super().__init__(**kwargs)
if TAVILY_AVAILABLE:
api_key = os.getenv("TAVILY_API_KEY")
self._client = TavilyClient(api_key=api_key)
self._async_client = AsyncTavilyClient(api_key=api_key)
self._client = TavilyClient(api_key=api_key, client_name="crewai")
self._async_client = AsyncTavilyClient(
api_key=api_key, client_name="crewai"
)
else:
try:
import subprocess

6
lib/crewai-tools/src/crewai_tools/tools/tavily_search_tool/tavily_search_tool.py
View File

@@ -115,9 +115,11 @@ class TavilySearchTool(BaseTool):
def __init__(self, **kwargs: Any):
super().__init__(**kwargs)
if TAVILY_AVAILABLE:
self.client = TavilyClient(api_key=self.api_key, proxies=self.proxies)
self.client = TavilyClient(
api_key=self.api_key, proxies=self.proxies, client_name="crewai"
)
self.async_client = AsyncTavilyClient(
api_key=self.api_key, proxies=self.proxies
api_key=self.api_key, proxies=self.proxies, client_name="crewai"
)
else:
try:

144
lib/crewai/src/crewai/experimental/agent_executor.py
View File

@@ -106,6 +106,7 @@ from crewai.utilities.planning_types import (
TodoItem,
TodoList,
)
from crewai.utilities.prompts import StandardPromptResult, SystemPromptResult
from crewai.utilities.step_execution_context import StepExecutionContext, StepResult
from crewai.utilities.string_utils import sanitize_tool_name
from crewai.utilities.tool_utils import execute_tool_and_check_finality
@@ -118,7 +119,6 @@ if TYPE_CHECKING:
from crewai.agents.tools_handler import ToolsHandler
from crewai.llms.base_llm import BaseLLM
from crewai.tools.tool_types import ToolResult
from crewai.utilities.prompts import StandardPromptResult, SystemPromptResult
_RouteT = TypeVar("_RouteT", bound=str)
@@ -218,6 +218,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
_instance_id: str = PrivateAttr(default_factory=lambda: str(uuid4())[:8])
_step_executor: Any = PrivateAttr(default=None)
_planner_observer: PlannerObserver | None = PrivateAttr(default=None)
_is_feedback_iteration: bool = PrivateAttr(default=False)
@model_validator(mode="after")
def _setup_executor(self) -> Self:
@@ -296,6 +297,33 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
"""Set state messages."""
self._state.messages = value
def _setup_messages(self, inputs: dict[str, Any]) -> None:
"""Set up messages for the agent execution."""
provider = get_provider()
if provider.setup_messages(cast("ExecutorContext", self)):
return
from crewai.llms.cache import mark_cache_breakpoint
if isinstance(self.prompt, SystemPromptResult):
system_prompt = self._format_prompt(self.prompt["system"], inputs)
user_prompt = self._format_prompt(self.prompt["user"], inputs)
self.state.messages.append(
mark_cache_breakpoint(
format_message_for_llm(system_prompt, role="system")
)
)
self.state.messages.append(
mark_cache_breakpoint(format_message_for_llm(user_prompt))
)
elif isinstance(self.prompt, StandardPromptResult):
user_prompt = self._format_prompt(self.prompt["prompt"], inputs)
self.state.messages.append(
mark_cache_breakpoint(format_message_for_llm(user_prompt))
)
provider.post_setup_messages(cast("ExecutorContext", self))
@property
def ask_for_human_input(self) -> bool:
"""Compatibility property - returns state ask_for_human_input."""
@@ -314,6 +342,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
enabled on the agent, it generates a plan before execution begins.
The plan is stored in state and todos are created from the steps.
"""
if self._is_feedback_iteration:
return
if not getattr(self.agent, "planning_enabled", False):
return
@@ -2761,27 +2791,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
"AgentExecutor.llm or .prompt is unset; the executor was "
"not fully restored or initialized before execution."
)
if "system" in self.prompt:
from crewai.llms.cache import mark_cache_breakpoint
prompt = cast("SystemPromptResult", self.prompt)
system_prompt = self._format_prompt(prompt["system"], inputs)
user_prompt = self._format_prompt(prompt["user"], inputs)
self.state.messages.append(
mark_cache_breakpoint(
format_message_for_llm(system_prompt, role="system")
)
)
self.state.messages.append(
mark_cache_breakpoint(format_message_for_llm(user_prompt))
)
else:
from crewai.llms.cache import mark_cache_breakpoint
user_prompt = self._format_prompt(self.prompt["prompt"], inputs)
self.state.messages.append(
mark_cache_breakpoint(format_message_for_llm(user_prompt))
)
self._setup_messages(inputs)
self._inject_files_from_inputs(inputs)
@@ -2867,27 +2877,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
"AgentExecutor.llm or .prompt is unset; the executor was "
"not fully restored or initialized before execution."
)
if "system" in self.prompt:
from crewai.llms.cache import mark_cache_breakpoint
prompt = cast("SystemPromptResult", self.prompt)
system_prompt = self._format_prompt(prompt["system"], inputs)
user_prompt = self._format_prompt(prompt["user"], inputs)
self.state.messages.append(
mark_cache_breakpoint(
format_message_for_llm(system_prompt, role="system")
)
)
self.state.messages.append(
mark_cache_breakpoint(format_message_for_llm(user_prompt))
)
else:
from crewai.llms.cache import mark_cache_breakpoint
user_prompt = self._format_prompt(self.prompt["prompt"], inputs)
self.state.messages.append(
mark_cache_breakpoint(format_message_for_llm(user_prompt))
)
self._setup_messages(inputs)
await self._ainject_files_from_inputs(inputs)
@@ -3169,8 +3159,13 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
Returns:
Final answer after feedback.
"""
self.messages = self.state.messages
provider = get_provider()
return provider.handle_feedback(formatted_answer, cast("ExecutorContext", self))
final_answer = provider.handle_feedback(
formatted_answer, cast("ExecutorContext", self)
)
self._complete_feedback(final_answer)
return final_answer
async def _ahandle_human_feedback(
self, formatted_answer: AgentFinish
@@ -3183,10 +3178,63 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
Returns:
Final answer after feedback.
"""
self.messages = self.state.messages
provider = get_provider()
return await provider.handle_feedback_async(
final_answer = await provider.handle_feedback_async(
formatted_answer, cast("AsyncExecutorContext", self)
)
self._complete_feedback(final_answer)
return final_answer
def _complete_feedback(self, final_answer: AgentFinish) -> None:
"""Mark the final reviewed answer as the completed executor state."""
self.state.current_answer = final_answer
self.state.is_finished = True
self.state.ask_for_human_input = False
self._finalize_called = True
def _prepare_feedback_iteration(self) -> None:
"""Reset flow completion state before rerunning with feedback."""
self._finalize_called = False
self._is_feedback_iteration = True
self.state.current_answer = None
self.state.is_finished = False
self.state.iterations = 0
self.state.use_native_tools = False
self.state.pending_tool_calls = []
self.state.plan = None
self.state.plan_ready = False
self.state.todos = TodoList()
self.state.replan_count = 0
self.state.last_replan_reason = None
self.state.observations = {}
self.state.execution_log = []
def _invoke_loop(self) -> AgentFinish:
"""Re-run the executor flow using the existing feedback messages."""
self._prepare_feedback_iteration()
try:
self.kickoff()
finally:
self._is_feedback_iteration = False
if not isinstance(self.state.current_answer, AgentFinish):
raise RuntimeError("Agent execution ended without reaching a final answer.")
return self.state.current_answer
async def _ainvoke_loop(self) -> AgentFinish:
"""Re-run the executor flow asynchronously using feedback messages."""
self._prepare_feedback_iteration()
try:
await self.kickoff_async()
finally:
self._is_feedback_iteration = False
if not isinstance(self.state.current_answer, AgentFinish):
raise RuntimeError("Agent execution ended without reaching a final answer.")
return self.state.current_answer
def _is_training_mode(self) -> bool:
"""Check if training mode is active.
@@ -3196,6 +3244,12 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
"""
return bool(self.crew and self.crew._train)
def _format_feedback_message(self, feedback: str) -> LLMMessage:
"""Format human feedback as an LLM message."""
return format_message_for_llm(
I18N_DEFAULT.slice("feedback_instructions").format(feedback=feedback)
)
# Backward compatibility alias (deprecated)
CrewAgentExecutorFlow = AgentExecutor

1
lib/crewai/src/crewai/telemetry/telemetry.py
View File

@@ -949,6 +949,7 @@ class Telemetry:
def _operation() -> None:
tracer = trace.get_tracer("crewai.telemetry")
span = tracer.start_span("Flow Creation")
self._add_attribute(span, "crewai_version", version("crewai"))
self._add_attribute(span, "flow_name", flow_name)
close_span(span)

94
lib/crewai/tests/agents/test_agent.py
View File

@@ -3,13 +3,14 @@
import os
import threading
from unittest import mock
from unittest.mock import MagicMock, patch
from unittest.mock import AsyncMock, MagicMock, patch
import warnings
from crewai.agents.crew_agent_executor import AgentFinish, CrewAgentExecutor
from crewai.constants import DEFAULT_LLM_MODEL
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.tool_usage_events import ToolUsageFinishedEvent
from crewai.experimental.agent_executor import AgentExecutor
from crewai.knowledge.knowledge import Knowledge
from crewai.knowledge.knowledge_config import KnowledgeConfig
from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
@@ -802,6 +803,97 @@ def test_agent_human_input():
assert output.strip().lower() == "hello"
def test_agent_default_executor_human_input():
from crewai.core.providers.human_input import SyncHumanInputProvider
agent = Agent(
role="test role",
goal="test goal",
backstory="test backstory",
)
task = Task(
agent=agent,
description="Say the word: Hi",
expected_output="The word: Hi",
human_input=True,
)
answers = iter(
[
AgentFinish(output="Hi", thought="", text="Hi"),
AgentFinish(output="Hello", thought="", text="Hello"),
]
)
feedback_responses = iter(["Don't say hi, say Hello instead!", ""])
def kickoff_side_effect(executor, *_args, **_kwargs):
executor.state.current_answer = next(answers)
executor.state.is_finished = True
with (
patch.object(
SyncHumanInputProvider,
"_prompt_input",
side_effect=lambda *_args, **_kwargs: next(feedback_responses),
) as mock_prompt_input,
patch.object(
AgentExecutor, "kickoff", autospec=True, side_effect=kickoff_side_effect
) as mock_kickoff,
):
output = agent.execute_task(task)
assert output == "Hello"
assert mock_prompt_input.call_count == 2
assert mock_kickoff.call_count == 2
@pytest.mark.asyncio
async def test_agent_default_executor_async_human_input():
from crewai.core.providers.human_input import SyncHumanInputProvider
agent = Agent(
role="test role",
goal="test goal",
backstory="test backstory",
)
task = Task(
agent=agent,
description="Say the word: Hi",
expected_output="The word: Hi",
human_input=True,
)
answers = iter(
[
AgentFinish(output="Hi", thought="", text="Hi"),
AgentFinish(output="Hello", thought="", text="Hello"),
]
)
feedback_responses = iter(["Don't say hi, say Hello instead!", ""])
async def kickoff_side_effect(executor, *_args, **_kwargs):
executor.state.current_answer = next(answers)
executor.state.is_finished = True
with (
patch.object(
SyncHumanInputProvider,
"_prompt_input_async",
new_callable=AsyncMock,
side_effect=lambda *_args, **_kwargs: next(feedback_responses),
) as mock_prompt_input,
patch.object(
AgentExecutor,
"kickoff_async",
autospec=True,
side_effect=kickoff_side_effect,
) as mock_kickoff,
):
output = await agent.aexecute_task(task)
assert output == "Hello"
assert mock_prompt_input.await_count == 2
assert mock_kickoff.await_count == 2
def test_interpolate_inputs():
agent = Agent(
role="{topic} specialist",

195
lib/crewai/tests/agents/test_agent_executor.py
View File

@@ -18,6 +18,7 @@ import pytest
from pydantic import BaseModel
from crewai.agents.tools_handler import ToolsHandler as _ToolsHandler
from crewai.core.providers.human_input import SyncHumanInputProvider
from crewai.agents.step_executor import StepExecutor
@@ -27,6 +28,13 @@ def _build_executor(**kwargs: Any) -> AgentExecutor:
Uses model_construct to skip Pydantic validators so plain Mock()
objects are accepted for typed fields like llm, agent, crew, task.
"""
prompt = kwargs.get("prompt")
if isinstance(prompt, dict):
if "system" in prompt:
kwargs["prompt"] = SystemPromptResult(**prompt)
else:
kwargs["prompt"] = StandardPromptResult(**prompt)
executor = AgentExecutor.model_construct(**kwargs)
executor._state = AgentExecutorState()
executor._methods = {}
@@ -50,6 +58,7 @@ def _build_executor(**kwargs: Any) -> AgentExecutor:
executor._last_context_error = None
executor._step_executor = None
executor._planner_observer = None
executor._is_feedback_iteration = False
return executor
from crewai.agents.planner_observer import PlannerObserver
from crewai.experimental.agent_executor import (
@@ -68,7 +77,8 @@ from crewai.events.types.tool_usage_events import (
)
from crewai.tools.tool_types import ToolResult
from crewai.utilities.step_execution_context import StepExecutionContext
from crewai.utilities.planning_types import TodoItem
from crewai.utilities.planning_types import TodoItem, TodoList
from crewai.utilities.prompts import StandardPromptResult, SystemPromptResult
from crewai.utilities.file_store import clear_files, clear_task_files, store_files
from crewai_files import TextFile
@@ -119,6 +129,189 @@ class TestAgentExecutor:
class StructuredResult(BaseModel):
value: str
def test_setup_messages_calls_human_input_provider_hooks(self):
"""Message setup should preserve the HumanInputProvider hook contract."""
executor = _build_executor(
prompt=StandardPromptResult(prompt="Original task: {input}"),
)
provider = Mock()
provider.setup_messages.return_value = False
def post_setup(context: AgentExecutor) -> None:
context.messages.append(
{"role": "system", "content": "provider post setup"}
)
provider.post_setup_messages.side_effect = post_setup
with patch(
"crewai.experimental.agent_executor.get_provider", return_value=provider
):
executor._setup_messages(
{"input": "draft this", "tool_names": "", "tools": ""}
)
provider.setup_messages.assert_called_once_with(executor)
provider.post_setup_messages.assert_called_once_with(executor)
assert executor.state.messages[0]["role"] == "user"
assert executor.state.messages[0]["content"] == "Original task: draft this"
assert executor.state.messages[1] == {
"role": "system",
"content": "provider post setup",
}
def test_setup_messages_can_be_owned_by_human_input_provider(self):
"""Providers can skip standard prompt setup by returning True."""
executor = _build_executor(
prompt=StandardPromptResult(prompt="Original task: {input}"),
)
provider = Mock()
def setup(context: AgentExecutor) -> bool:
context.messages.append({"role": "user", "content": "provider message"})
return True
provider.setup_messages.side_effect = setup
with patch(
"crewai.experimental.agent_executor.get_provider", return_value=provider
):
executor._setup_messages(
{"input": "draft this", "tool_names": "", "tools": ""}
)
provider.setup_messages.assert_called_once_with(executor)
provider.post_setup_messages.assert_not_called()
assert executor.state.messages == [
{"role": "user", "content": "provider message"}
]
def test_human_feedback_reruns_flow_with_state_messages(self):
"""Human feedback should use AgentExecutor state messages."""
executor = _build_executor(agent=SimpleNamespace(verbose=False), crew=None)
executor.state.messages = [{"role": "user", "content": "original task"}]
executor.state.current_answer = AgentFinish(
thought="", output="draft", text="draft"
)
executor.state.is_finished = True
executor._finalize_called = True
executor.ask_for_human_input = True
executor.state.iterations = executor.max_iter
executor.state.plan = "completed plan"
executor.state.plan_ready = True
executor.state.todos = TodoList(
items=[TodoItem(step_number=1, description="Done", status="completed")]
)
improved_answer = AgentFinish(thought="", output="improved", text="improved")
feedback_responses = iter(["make it friendlier", ""])
def finish_feedback_iteration(*_args: Any, **_kwargs: Any) -> None:
assert executor._is_feedback_iteration is True
assert executor.state.iterations == 0
assert executor.state.plan is None
assert executor.state.todos.items == []
executor.state.current_answer = improved_answer
executor.state.is_finished = True
with (
patch.object(
SyncHumanInputProvider,
"_prompt_input",
side_effect=lambda *_args, **_kwargs: next(feedback_responses),
) as mock_prompt_input,
patch.object(
AgentExecutor, "kickoff", side_effect=finish_feedback_iteration
) as mock_kickoff,
):
result = executor._handle_human_feedback(
AgentFinish(thought="", output="draft", text="draft")
)
assert result is improved_answer
assert mock_prompt_input.call_count == 2
mock_kickoff.assert_called_once()
assert executor.messages is executor.state.messages
assert "make it friendlier" in executor.state.messages[-1]["content"]
assert executor.ask_for_human_input is False
assert executor.state.current_answer is improved_answer
assert executor.state.is_finished is True
assert executor._finalize_called is True
assert executor._is_feedback_iteration is False
@pytest.mark.asyncio
async def test_async_human_feedback_reruns_flow_with_state_messages(self):
"""Async human feedback should use AgentExecutor state messages."""
executor = _build_executor(agent=SimpleNamespace(verbose=False), crew=None)
executor.state.messages = [{"role": "user", "content": "original task"}]
executor.state.current_answer = AgentFinish(
thought="", output="draft", text="draft"
)
executor.state.is_finished = True
executor._finalize_called = True
executor.ask_for_human_input = True
executor.state.iterations = executor.max_iter
executor.state.plan = "completed plan"
executor.state.plan_ready = True
executor.state.todos = TodoList(
items=[TodoItem(step_number=1, description="Done", status="completed")]
)
improved_answer = AgentFinish(thought="", output="improved", text="improved")
feedback_responses = iter(["make it friendlier", ""])
async def finish_feedback_iteration(*_args: Any, **_kwargs: Any) -> None:
assert executor._is_feedback_iteration is True
assert executor.state.iterations == 0
assert executor.state.plan is None
assert executor.state.todos.items == []
executor.state.current_answer = improved_answer
executor.state.is_finished = True
with (
patch.object(
SyncHumanInputProvider,
"_prompt_input_async",
new_callable=AsyncMock,
side_effect=lambda *_args, **_kwargs: next(feedback_responses),
) as mock_prompt_input,
patch.object(
AgentExecutor,
"kickoff_async",
new_callable=AsyncMock,
side_effect=finish_feedback_iteration,
) as mock_kickoff,
):
result = await executor._ahandle_human_feedback(
AgentFinish(thought="", output="draft", text="draft")
)
assert result is improved_answer
assert mock_prompt_input.await_count == 2
mock_kickoff.assert_awaited_once()
assert executor.messages is executor.state.messages
assert "make it friendlier" in executor.state.messages[-1]["content"]
assert executor.ask_for_human_input is False
assert executor.state.current_answer is improved_answer
assert executor.state.is_finished is True
assert executor._finalize_called is True
assert executor._is_feedback_iteration is False
def test_feedback_iteration_skips_plan_generation(self):
"""Feedback reruns should reason over feedback without regenerating a plan."""
executor = _build_executor(
agent=SimpleNamespace(planning_enabled=True, verbose=False),
task=SimpleNamespace(),
)
executor._is_feedback_iteration = True
with patch("crewai.utilities.reasoning_handler.AgentReasoning") as reasoning:
executor.generate_plan()
reasoning.assert_not_called()
assert executor.state.plan is None
assert executor.state.todos.items == []
def test_inject_files_from_crew_task_store(self):
"""Crew-level input_files should attach to the LLM user message."""
crew_id = uuid4()

26
lib/crewai/tests/telemetry/test_telemetry.py
View File

@@ -96,6 +96,32 @@ def test_flow_execution_span_records_crewai_version():
span.set_attribute.assert_any_call("flow_name", "ResearchFlow")
def test_flow_creation_span_records_crewai_version():
tracer = Mock()
span = Mock()
tracer.start_span.return_value = span
with (
patch.dict(
os.environ,
{
"CREWAI_DISABLE_TELEMETRY": "false",
"CREWAI_DISABLE_TRACKING": "false",
"OTEL_SDK_DISABLED": "false",
},
),
patch("crewai.telemetry.telemetry.TracerProvider"),
patch("crewai.telemetry.telemetry.trace.get_tracer", return_value=tracer),
patch("crewai.telemetry.telemetry.version", return_value="9.9.9"),
):
telemetry = Telemetry()
telemetry.flow_creation_span("ResearchFlow")
tracer.start_span.assert_called_once_with("Flow Creation")
span.set_attribute.assert_any_call("crewai_version", "9.9.9")
span.set_attribute.assert_any_call("flow_name", "ResearchFlow")
@patch("crewai.telemetry.telemetry.logger.error")
@patch(
"opentelemetry.exporter.otlp.proto.http.trace_exporter.OTLPSpanExporter.export",

6
lib/crewai/tests/test_crew.py
View File

@@ -2908,12 +2908,6 @@ def test_manager_agent_with_tools_raises_exception(researcher, writer):
crew.kickoff()
@pytest.mark.xfail(
strict=True,
reason="crew.train() relies on CrewAgentExecutor._format_feedback_message; "
"AgentExecutor (the new default) does not implement training feedback yet. "
"Remove this xfail once training is migrated to AgentExecutor.",
)
@pytest.mark.vcr()
def test_crew_train_success(researcher, writer, monkeypatch):
task = Task(

25
lib/crewai/tests/test_flow_human_input_integration.py
View File

@@ -138,4 +138,27 @@ class TestFlowHumanInputIntegration:
for call in call_args
if call[0]
)
assert training_panel_found
assert training_panel_found
@patch("builtins.input", return_value="please make it warmer")
def test_non_empty_input_prints_processing_feedback(self, mock_input):
"""Non-empty input should be displayed as feedback to process."""
provider = SyncHumanInputProvider()
crew = MagicMock()
crew._train = False
formatter = event_listener.formatter
with (
patch.object(formatter, "pause_live_updates"),
patch.object(formatter, "resume_live_updates"),
patch.object(formatter.console, "print") as mock_console_print,
):
result = provider._prompt_input(crew)
assert result == "please make it warmer"
mock_input.assert_called_once()
printed_text = "\n".join(
str(call.args[0]) for call in mock_console_print.call_args_list
)
assert "Processing your feedback" in printed_text