feat(cli): improve platform integration setup UX (#7453)

* fix(cli): silence tool import warnings during platform setup

* feat(cli): validate platform integrations concurrently

* fix(cli): persist platform token during crew setup

* fix(core): make settings write probe concurrency-safe

* test(cli): simplify event loop fallback stub

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(

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

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

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