crewAI/lib/crewai/src/crewai/cli/command.py

import json

import httpx
from rich.console import Console

from crewai.cli.authentication.token import get_auth_token
from crewai.cli.plus_api import PlusAPI
from crewai.telemetry.telemetry import Telemetry


console = Console()


class BaseCommand:
    def __init__(self) -> None:
        self._telemetry = Telemetry()
        self._telemetry.set_tracer()


class PlusAPIMixin:
    def __init__(self, telemetry: Telemetry) -> None:
        try:
            telemetry.set_tracer()
            self.plus_api_client = PlusAPI(api_key=get_auth_token())
        except Exception:
            telemetry.deploy_signup_error_span()
            console.print(
                "Please sign up/login to CrewAI+ before using the CLI.",
                style="bold red",
            )
            console.print("Run 'crewai login' to sign up/login.", style="bold green")
            raise SystemExit from None

    def _validate_response(self, response: httpx.Response) -> None:
        """
        Handle and display error messages from API responses.

        Args:
            response (httpx.Response): The response from the Plus API
        """
        try:
            json_response = response.json()
        except (json.JSONDecodeError, ValueError):
            console.print(
                "Failed to parse response from Enterprise API failed. Details:",
                style="bold red",
            )
            console.print(f"Status Code: {response.status_code}")
            console.print(
                f"Response:\n{response.content.decode('utf-8', errors='replace')}"
            )
            raise SystemExit from None

        if response.status_code == 422:
            console.print(
                "Failed to complete operation. Please fix the following errors:",
                style="bold red",
            )
            for field, messages in json_response.items():
                for message in messages:
                    console.print(
                        f"* [bold red]{field.capitalize()}[/bold red] {message}"
                    )
            raise SystemExit

        if not response.is_success:
            console.print(
                "Request to Enterprise API failed. Details:", style="bold red"
            )
            details = (
                json_response.get("error")
                or json_response.get("message")
                or response.content.decode("utf-8", errors="replace")
            )
            console.print(f"{details}")
            raise SystemExit