chore: apply ruff linting fixes to CLI module

fix: apply ruff fixes to CLI and update Okta provider test

Author: greysonlalonde

src/crewai/cli/authentication/providers/__init__.py

@@ -1,5 +1,6 @@
 from crewai.cli.authentication.providers.base_provider import BaseProvider
 
+
 class Auth0Provider(BaseProvider):
     def get_authorize_url(self) -> str:
         return f"https://{self._get_domain()}/oauth/device/code"
@@ -14,13 +15,20 @@ def get_issuer(self) -> str:
         return f"https://{self._get_domain()}/"
 
     def get_audience(self) -> str:
-        assert self.settings.audience is not None, "Audience is required"
+        if self.settings.audience is None:
+            raise ValueError(
+                "Audience is required. Please set it in the configuration."
+            )
         return self.settings.audience
 
     def get_client_id(self) -> str:
-        assert self.settings.client_id is not None, "Client ID is required"
+        if self.settings.client_id is None:
+            raise ValueError(
+                "Client ID is required. Please set it in the configuration."
+            )
         return self.settings.client_id
 
     def _get_domain(self) -> str:
-        assert self.settings.domain is not None, "Domain is required"
+        if self.settings.domain is None:
+            raise ValueError("Domain is required. Please set it in the configuration.")
         return self.settings.domain

src/crewai/cli/authentication/providers/auth0.py

@@ -1,30 +1,26 @@
 from abc import ABC, abstractmethod
+
 from crewai.cli.authentication.main import Oauth2Settings
 
+
 class BaseProvider(ABC):
     def __init__(self, settings: Oauth2Settings):
         self.settings = settings
 
     @abstractmethod
-    def get_authorize_url(self) -> str:
-        ...
+    def get_authorize_url(self) -> str: ...
 
     @abstractmethod
-    def get_token_url(self) -> str:
-        ...
+    def get_token_url(self) -> str: ...
 
     @abstractmethod
-    def get_jwks_url(self) -> str:
-        ...
+    def get_jwks_url(self) -> str: ...
 
     @abstractmethod
-    def get_issuer(self) -> str:
-        ...
+    def get_issuer(self) -> str: ...
 
     @abstractmethod
-    def get_audience(self) -> str:
-        ...
+    def get_audience(self) -> str: ...
 
     @abstractmethod
-    def get_client_id(self) -> str:
-        ...
+    def get_client_id(self) -> str: ...

src/crewai/cli/authentication/providers/base_provider.py

@@ -1,5 +1,6 @@
 from crewai.cli.authentication.providers.base_provider import BaseProvider
 
+
 class OktaProvider(BaseProvider):
     def get_authorize_url(self) -> str:
         return f"https://{self.settings.domain}/oauth2/default/v1/device/authorize"
@@ -14,9 +15,15 @@ def get_issuer(self) -> str:
         return f"https://{self.settings.domain}/oauth2/default"
 
     def get_audience(self) -> str:
-        assert self.settings.audience is not None
+        if self.settings.audience is None:
+            raise ValueError(
+                "Audience is required. Please set it in the configuration."
+            )
         return self.settings.audience
 
     def get_client_id(self) -> str:
-        assert self.settings.client_id is not None
+        if self.settings.client_id is None:
+            raise ValueError(
+                "Client ID is required. Please set it in the configuration."
+            )
         return self.settings.client_id

src/crewai/cli/authentication/providers/okta.py

@@ -1,5 +1,6 @@
 from crewai.cli.authentication.providers.base_provider import BaseProvider
 
+
 class WorkosProvider(BaseProvider):
     def get_authorize_url(self) -> str:
         return f"https://{self._get_domain()}/oauth2/device_authorization"
@@ -17,9 +18,13 @@ def get_audience(self) -> str:
         return self.settings.audience or ""
 
     def get_client_id(self) -> str:
-        assert self.settings.client_id is not None, "Client ID is required"
+        if self.settings.client_id is None:
+            raise ValueError(
+                "Client ID is required. Please set it in the configuration."
+            )
         return self.settings.client_id
 
     def _get_domain(self) -> str:
-        assert self.settings.domain is not None, "Domain is required"
+        if self.settings.domain is None:
+            raise ValueError("Domain is required. Please set it in the configuration.")
         return self.settings.domain

src/crewai/cli/authentication/providers/workos.py

@@ -17,16 +17,14 @@ def validate_jwt_token(
                        missing required claims).
     """
 
-    decoded_token = None
-
     try:
         jwk_client = PyJWKClient(jwks_url)
         signing_key = jwk_client.get_signing_key_from_jwt(jwt_token)
 
         _unverified_decoded_token = jwt.decode(
             jwt_token, options={"verify_signature": False}
         )
-        decoded_token = jwt.decode(
+        return jwt.decode(
             jwt_token,
             signing_key.key,
             algorithms=["RS256"],
@@ -40,23 +38,22 @@ def validate_jwt_token(
                 "require": ["exp", "iat", "iss", "aud", "sub"],
             },
         )
-        return decoded_token
 
-    except jwt.ExpiredSignatureError:
-        raise Exception("Token has expired.")
-    except jwt.InvalidAudienceError:
+    except jwt.ExpiredSignatureError as e:
+        raise Exception("Token has expired.") from e
+    except jwt.InvalidAudienceError as e:
         actual_audience = _unverified_decoded_token.get("aud", "[no audience found]")
         raise Exception(
             f"Invalid token audience. Got: '{actual_audience}'. Expected: '{audience}'"
-        )
-    except jwt.InvalidIssuerError:
+        ) from e
+    except jwt.InvalidIssuerError as e:
         actual_issuer = _unverified_decoded_token.get("iss", "[no issuer found]")
         raise Exception(
             f"Invalid token issuer. Got: '{actual_issuer}'. Expected: '{issuer}'"
-        )
+        ) from e
     except jwt.MissingRequiredClaimError as e:
-        raise Exception(f"Token is missing required claims: {str(e)}")
+        raise Exception(f"Token is missing required claims: {e!s}") from e
     except jwt.exceptions.PyJWKClientError as e:
-        raise Exception(f"JWKS or key processing error: {str(e)}")
+        raise Exception(f"JWKS or key processing error: {e!s}") from e
     except jwt.InvalidTokenError as e:
-        raise Exception(f"Invalid token: {str(e)}")
+        raise Exception(f"Invalid token: {e!s}") from e

src/crewai/cli/authentication/utils.py

@@ -1,13 +1,13 @@
 from importlib.metadata import version as get_version
-from typing import Optional
 
 import click
-from crewai.cli.config import Settings
-from crewai.cli.settings.main import SettingsCommand
+
 from crewai.cli.add_crew_to_flow import add_crew_to_flow
+from crewai.cli.config import Settings
 from crewai.cli.create_crew import create_crew
 from crewai.cli.create_flow import create_flow
 from crewai.cli.crew_chat import run_chat
+from crewai.cli.settings.main import SettingsCommand
 from crewai.memory.storage.kickoff_task_outputs_storage import (
     KickoffTaskOutputsSQLiteStorage,
 )
@@ -237,13 +237,11 @@ def login():
 @crewai.group()
 def deploy():
     """Deploy the Crew CLI group."""
-    pass
 
 
 @crewai.group()
 def tool():
     """Tool Repository related commands."""
-    pass
 
 
 @deploy.command(name="create")
@@ -263,31 +261,31 @@ def deploy_list():
 
 @deploy.command(name="push")
 @click.option("-u", "--uuid", type=str, help="Crew UUID parameter")
-def deploy_push(uuid: Optional[str]):
+def deploy_push(uuid: str | None):
     """Deploy the Crew."""
     deploy_cmd = DeployCommand()
     deploy_cmd.deploy(uuid=uuid)
 
 
 @deploy.command(name="status")
 @click.option("-u", "--uuid", type=str, help="Crew UUID parameter")
-def deply_status(uuid: Optional[str]):
+def deply_status(uuid: str | None):
     """Get the status of a deployment."""
     deploy_cmd = DeployCommand()
     deploy_cmd.get_crew_status(uuid=uuid)
 
 
 @deploy.command(name="logs")
 @click.option("-u", "--uuid", type=str, help="Crew UUID parameter")
-def deploy_logs(uuid: Optional[str]):
+def deploy_logs(uuid: str | None):
     """Get the logs of a deployment."""
     deploy_cmd = DeployCommand()
     deploy_cmd.get_crew_logs(uuid=uuid)
 
 
 @deploy.command(name="remove")
 @click.option("-u", "--uuid", type=str, help="Crew UUID parameter")
-def deploy_remove(uuid: Optional[str]):
+def deploy_remove(uuid: str | None):
     """Remove a deployment."""
     deploy_cmd = DeployCommand()
     deploy_cmd.remove_crew(uuid=uuid)
@@ -327,7 +325,6 @@ def tool_publish(is_public: bool, force: bool):
 @crewai.group()
 def flow():
     """Flow related commands."""
-    pass
 
 
 @flow.command(name="kickoff")
@@ -359,7 +356,7 @@ def chat():
     and using the Chat LLM to generate responses.
     """
     click.secho(
-        "\nStarting a conversation with the Crew\n" "Type 'exit' or Ctrl+C to quit.\n",
+        "\nStarting a conversation with the Crew\nType 'exit' or Ctrl+C to quit.\n",
     )
 
     run_chat()
@@ -368,7 +365,6 @@ def chat():
 @crewai.group(invoke_without_command=True)
 def org():
     """Organization management commands."""
-    pass
 
 
 @org.command("list")
@@ -396,7 +392,6 @@ def current():
 @crewai.group()
 def enterprise():
     """Enterprise Configuration commands."""
-    pass
 
 
 @enterprise.command("configure")
@@ -410,7 +405,6 @@ def enterprise_configure(enterprise_url: str):
 @crewai.group()
 def config():
     """CLI Configuration commands."""
-    pass
 
 
 @config.command("list")

src/crewai/cli/cli.py

@@ -1,15 +1,14 @@
 import json
 from pathlib import Path
-from typing import Optional
 
 from pydantic import BaseModel, Field
 
 from crewai.cli.constants import (
-    DEFAULT_CREWAI_ENTERPRISE_URL,
-    CREWAI_ENTERPRISE_DEFAULT_OAUTH2_PROVIDER,
     CREWAI_ENTERPRISE_DEFAULT_OAUTH2_AUDIENCE,
     CREWAI_ENTERPRISE_DEFAULT_OAUTH2_CLIENT_ID,
     CREWAI_ENTERPRISE_DEFAULT_OAUTH2_DOMAIN,
+    CREWAI_ENTERPRISE_DEFAULT_OAUTH2_PROVIDER,
+    DEFAULT_CREWAI_ENTERPRISE_URL,
 )
 from crewai.cli.shared.token_manager import TokenManager
 
@@ -56,20 +55,20 @@
 
 
 class Settings(BaseModel):
-    enterprise_base_url: Optional[str] = Field(
+    enterprise_base_url: str | None = Field(
         default=DEFAULT_CLI_SETTINGS["enterprise_base_url"],
         description="Base URL of the CrewAI Enterprise instance",
     )
-    tool_repository_username: Optional[str] = Field(
+    tool_repository_username: str | None = Field(
         None, description="Username for interacting with the Tool Repository"
     )
-    tool_repository_password: Optional[str] = Field(
+    tool_repository_password: str | None = Field(
         None, description="Password for interacting with the Tool Repository"
     )
-    org_name: Optional[str] = Field(
+    org_name: str | None = Field(
         None, description="Name of the currently active organization"
     )
-    org_uuid: Optional[str] = Field(
+    org_uuid: str | None = Field(
         None, description="UUID of the currently active organization"
     )
     config_path: Path = Field(default=DEFAULT_CONFIG_PATH, frozen=True, exclude=True)
@@ -79,7 +78,7 @@ class Settings(BaseModel):
         default=DEFAULT_CLI_SETTINGS["oauth2_provider"],
     )
 
-    oauth2_audience: Optional[str] = Field(
+    oauth2_audience: str | None = Field(
         description="OAuth2 audience value, typically used to identify the target API or resource.",
         default=DEFAULT_CLI_SETTINGS["oauth2_audience"],
     )