diff --git a/lib/cli/src/crewai_cli/create_json_crew.py b/lib/cli/src/crewai_cli/create_json_crew.py index 1586ba52a..e49562fcf 100644 --- a/lib/cli/src/crewai_cli/create_json_crew.py +++ b/lib/cli/src/crewai_cli/create_json_crew.py @@ -2,12 +2,14 @@ from __future__ import annotations +import asyncio import json import os from pathlib import Path import re import sys from typing import Any +import warnings import click from crewai_core.telemetry import Telemetry @@ -622,10 +624,6 @@ def _wizard_agents_and_tasks( "inputs": {}, } - # Platform authentication belongs to the final wizard step, after the - # user has finished configuring agents, tasks, and crew settings. - _setup_platform_auth(agents) - return agents, tasks, crew_settings @@ -919,48 +917,106 @@ def _prompt_platform_token() -> str: ).strip() +def _check_platform_app( + app: str, application_selector: Any, client_for_selector: Any +) -> tuple[Any | None, Exception | None]: + """Check one AMP application with the synchronous platform client.""" + try: + selector = application_selector.from_string(app) + return client_for_selector(selector).get_actions([selector]), None + except Exception as error: + return None, error + + +async def _check_platform_apps_concurrently( + apps: list[str], application_selector: Any, client_for_selector: Any +) -> list[tuple[Any | None, Exception | None]]: + """Run independent AMP application checks concurrently.""" + return await asyncio.gather( + *( + asyncio.to_thread( + _check_platform_app, app, application_selector, client_for_selector + ) + for app in apps + ) + ) + + +def _report_platform_app_validation( + app: str, + actions: Any | None, + error: Exception | None, + failed: list[str], +) -> bool: + """Print one AMP application validation result and return token validity.""" + app_name = _platform_app_name(app) + if error is not None: + status_code = getattr(getattr(error, "response", None), "status_code", None) + if status_code in {401, 403}: + click.secho( + " ✘ CrewAI Platform Integration Token is invalid or expired", + fg="red", + ) + return True + click.secho( + f" ✘ {app_name} integration could not be validated: {error}", + fg="red", + ) + failed.append(app) + return False + + if not actions: + click.secho( + f" ✘ {app_name} integration is not connected on CrewAI Platform", + fg="red", + ) + failed.append(app) + else: + click.secho( + f" ✔ {app_name} integration is connected on CrewAI Platform", + fg="green", + ) + return False + + def _validate_platform_apps( apps: list[str], application_selector: Any, client_for_selector: Any ) -> tuple[list[str], bool]: - """Check selected AMP applications and return failures and token validity.""" + """Check selected AMP applications concurrently when no event loop is running.""" failed: list[str] = [] - for app in apps: - app_name = _platform_app_name(app) + try: + asyncio.get_running_loop() + except RuntimeError: + app_names = [_platform_app_name(app) for app in apps] click.echo() click.secho( - " Checking CrewAI Platform Integration Token and " - f"{app_name} integration on AMP...", + " Checking " + f"{', '.join(app_names)} integration{'s' if len(app_names) != 1 else ''} " + "together on AMP...", fg="cyan", ) - try: - selector = application_selector.from_string(app) - actions = client_for_selector(selector).get_actions([selector]) - except Exception as error: - status_code = getattr(getattr(error, "response", None), "status_code", None) - if status_code in {401, 403}: - click.secho( - " ✘ CrewAI Platform Integration Token is invalid or expired", - fg="red", - ) + results = asyncio.run( + _check_platform_apps_concurrently( + apps, application_selector, client_for_selector + ) + ) + for app, (actions, error) in zip(apps, results, strict=True): + if _report_platform_app_validation(app, actions, error, failed): return failed, True + else: + for app in apps: + app_name = _platform_app_name(app) + click.echo() click.secho( - f" ✘ {app_name} integration could not be validated: {error}", - fg="red", + " Checking CrewAI Platform Integration Token and " + f"{app_name} integration on AMP...", + fg="cyan", ) - failed.append(app) - continue - - if not actions: - click.secho( - f" ✘ {app_name} integration is not connected on CrewAI Platform", - fg="red", - ) - failed.append(app) - else: - click.secho( - f" ✔ {app_name} integration is connected on CrewAI Platform", - fg="green", + actions, error = _check_platform_app( + app, application_selector, client_for_selector ) + if _report_platform_app_validation(app, actions, error, failed): + return failed, True return failed, False @@ -1011,10 +1067,14 @@ def _setup_platform_auth(agents: list[dict[str, Any]]) -> str | None: click.echo() try: - from crewai_tools.tools.crewai_platform_tools.integrations_client import ( - ApplicationSelector, - client_for_selector, - ) + # Importing crewai_tools currently initializes optional tool SDKs. Keep + # their import-time warnings out of the interactive token setup flow. + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + from crewai_tools.tools.crewai_platform_tools.integrations_client import ( + ApplicationSelector, + client_for_selector, + ) except ImportError as error: raise click.ClickException( "Platform tools require the 'crewai-tools' package. " @@ -1105,11 +1165,9 @@ def create_json_crew( default_llm=default_llm, ) - platform_token = ( - os.environ.get("CREWAI_PLATFORM_INTEGRATION_TOKEN") - if _platform_apps_from_agents(agents) and not dmn_mode - else None - ) + # Authenticate only after the full wizard is complete, but before any + # project files are created. The returned token is then persisted below. + platform_token = _setup_platform_auth(agents) if not dmn_mode else None # Create directories only after platform authentication succeeds. folder_path.mkdir(parents=True) @@ -1124,6 +1182,7 @@ def create_json_crew( env_vars = load_env_vars(folder_path) env_vars["CREWAI_PLATFORM_INTEGRATION_TOKEN"] = platform_token write_env_file(folder_path, env_vars) + _success("CrewAI Platform integration token saved to .env") for agent in agents: _write_jsonc( diff --git a/lib/cli/tests/test_create_crew.py b/lib/cli/tests/test_create_crew.py index f173fed80..95f1ba1c0 100644 --- a/lib/cli/tests/test_create_crew.py +++ b/lib/cli/tests/test_create_crew.py @@ -1,6 +1,10 @@ +import asyncio +import builtins import keyword import shutil import tempfile +import threading +import warnings from pathlib import Path from unittest import mock @@ -667,6 +671,118 @@ def test_json_wizard_platform_catalog_contains_every_supported_app(): ] +def test_platform_auth_suppresses_warnings_only_while_importing_tools( + monkeypatch, +): + class FakeApplicationSelector: + @classmethod + def from_string(cls, value: str) -> str: + return value + + fake_client = mock.Mock() + fake_client.get_actions.return_value = [object()] + fake_integrations_client = mock.Mock( + ApplicationSelector=FakeApplicationSelector, + client_for_selector=lambda _selector: fake_client, + ) + original_import = builtins.__import__ + + def import_with_warning(name, globals=None, locals=None, fromlist=(), level=0): + if name == "crewai_tools.tools.crewai_platform_tools.integrations_client": + warnings.warn("optional dependency import warning", UserWarning) + return fake_integrations_client + return original_import(name, globals, locals, fromlist, level) + + monkeypatch.setattr(builtins, "__import__", import_with_warning) + monkeypatch.setenv("CREWAI_PLATFORM_INTEGRATION_TOKEN", "test-token") + + with warnings.catch_warnings(record=True) as caught_warnings: + warnings.simplefilter("always") + json_crew._setup_platform_auth([{"tools": ["platform:github"]}]) + warnings.warn("warning after import", UserWarning) + + assert [str(warning.message) for warning in caught_warnings] == [ + "warning after import" + ] + + +def test_platform_validation_checks_apps_concurrently(): + class FakeApplicationSelector: + @classmethod + def from_string(cls, value: str) -> str: + return value + + barrier = threading.Barrier(2) + + class FakeClient: + def get_actions(self, _selectors): + barrier.wait(timeout=1) + return [object()] + + results = asyncio.run( + json_crew._check_platform_apps_concurrently( + ["github", "gmail"], + FakeApplicationSelector, + lambda _selector: FakeClient(), + ) + ) + + assert all(actions and error is None for actions, error in results) + + +def test_platform_validation_announces_concurrent_apps_together(capsys): + class FakeApplicationSelector: + @classmethod + def from_string(cls, value: str) -> str: + return value + + class FakeClient: + def get_actions(self, _selectors): + return [object()] + + failed_apps, token_invalid = json_crew._validate_platform_apps( + ["github", "gmail", "google_calendar"], + FakeApplicationSelector, + lambda _selector: FakeClient(), + ) + + output = capsys.readouterr().out + assert (failed_apps, token_invalid) == ([], False) + assert "Checking GitHub, Gmail, Google Calendar integrations together on AMP" in output + assert "Checking CrewAI Platform Integration Token and GitHub" not in output + + +def test_platform_validation_falls_back_to_sequential_checks(monkeypatch, capsys): + class FakeApplicationSelector: + @classmethod + def from_string(cls, value: str) -> str: + return value + + checked_apps: list[str] = [] + + class FakeClient: + def __init__(self, app: str): + self.app = app + + def get_actions(self, _selectors): + checked_apps.append(self.app) + return [object()] + + monkeypatch.setattr(json_crew.asyncio, "get_running_loop", object) + + failed_apps, token_invalid = json_crew._validate_platform_apps( + ["github", "gmail"], + FakeApplicationSelector, + lambda app: FakeClient(app), + ) + + assert (failed_apps, token_invalid) == ([], False) + assert checked_apps == ["github", "gmail"] + output = capsys.readouterr().out + assert "Checking CrewAI Platform Integration Token and GitHub integration" in output + assert "Checking CrewAI Platform Integration Token and Gmail integration" in output + + def test_multi_picker_skips_separator_on_initial_cursor(monkeypatch): cursors: list[int] = [] @@ -889,6 +1005,44 @@ def test_json_create_provider_preselects_default_model(tmp_path, monkeypatch): assert '"knowledge_sources": []' in agent_template +def test_json_create_saves_platform_token_to_env_file(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + monkeypatch.setattr( + json_crew, + "_wizard_agents_and_tasks", + lambda **_: ( + [ + { + "name": "researcher", + "role": "Researcher", + "goal": "Research", + "backstory": "Researcher", + "llm": "openai/gpt-5.5", + "tools": ["platform:github"], + "planning": False, + "allow_delegation": False, + } + ], + [ + { + "name": "research_task", + "description": "Research", + "expected_output": "Findings", + "agent": "researcher", + "context": [], + } + ], + {"process": "sequential", "memory": False, "inputs": {}}, + ), + ) + monkeypatch.setattr(json_crew, "_setup_platform_auth", lambda _agents: "token") + + json_crew.create_json_crew("Platform Crew", skip_provider=True) + + env_file = tmp_path / "platform_crew" / ".env" + assert "CREWAI_PLATFORM_INTEGRATION_TOKEN=token" in env_file.read_text() + + def test_json_crew_uses_template_files(): template_names = { "pyproject.toml", diff --git a/lib/crewai-core/src/crewai_core/settings.py b/lib/crewai-core/src/crewai_core/settings.py index 967aa445c..be74957cf 100644 --- a/lib/crewai-core/src/crewai_core/settings.py +++ b/lib/crewai-core/src/crewai_core/settings.py @@ -92,14 +92,26 @@ def get_writable_config_path() -> Path | None: try: config_path.parent.mkdir(parents=True, exist_ok=True) _ensure_dir_mode(config_path.parent) - test_file = config_path.parent / ".crewai_write_test" + test_file: Path | None = None + fd: int | None = None try: - test_file.write_text("test") + fd, test_file_path = tempfile.mkstemp( + dir=config_path.parent, + prefix=".crewai_write_test.", + ) + test_file = Path(test_file_path) + os.close(fd) + fd = None test_file.unlink() logger.info(f"Using config path: {config_path}") return config_path except Exception: # noqa: S112 continue + finally: + if fd is not None: + os.close(fd) + if test_file is not None: + test_file.unlink(missing_ok=True) except Exception: # noqa: S112 continue diff --git a/lib/crewai-core/tests/test_settings.py b/lib/crewai-core/tests/test_settings.py new file mode 100644 index 000000000..6aa27d575 --- /dev/null +++ b/lib/crewai-core/tests/test_settings.py @@ -0,0 +1,38 @@ +"""Tests for CrewAI Core settings.""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +import json +from pathlib import Path +import threading + +import crewai_core.settings as settings_module +import pytest + + +def test_concurrent_settings_instances_keep_configured_enterprise_url( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Concurrent settings probes must not make a worker use fallback settings.""" + config_path = tmp_path / "config" / "settings.json" + config_path.parent.mkdir() + enterprise_url = "https://enterprise.example.com" + config_path.write_text(json.dumps({"enterprise_base_url": enterprise_url})) + monkeypatch.setattr(settings_module, "DEFAULT_CONFIG_PATH", config_path) + + original_unlink = Path.unlink + shared_probe_barrier = threading.Barrier(2) + + def synchronize_shared_probe_unlink(path: Path, missing_ok: bool = False) -> None: + if path.name == ".crewai_write_test": + shared_probe_barrier.wait(timeout=1) + original_unlink(path, missing_ok=missing_ok) + + monkeypatch.setattr(Path, "unlink", synchronize_shared_probe_unlink) + + with ThreadPoolExecutor(max_workers=2) as executor: + settings = list(executor.map(lambda _: settings_module.Settings(), range(2))) + + assert [item.enterprise_base_url for item in settings] == [enterprise_url] * 2 + assert not list(config_path.parent.glob(".crewai_write_test.*"))