A class to handle deployment-related operations for CrewAI projects.
| 167 | |
| 168 | |
| 169 | class DeployCommand(BaseCommand, PlusAPIMixin): |
| 170 | """ |
| 171 | A class to handle deployment-related operations for CrewAI projects. |
| 172 | """ |
| 173 | |
| 174 | def __init__(self) -> None: |
| 175 | """ |
| 176 | Initialize the DeployCommand with project name and API client. |
| 177 | """ |
| 178 | |
| 179 | BaseCommand.__init__(self) |
| 180 | PlusAPIMixin.__init__(self, telemetry=self._telemetry) |
| 181 | self.project_name = get_project_name(require=True) |
| 182 | |
| 183 | def _standard_no_param_error_message(self) -> None: |
| 184 | """ |
| 185 | Display a standard error message when no UUID or project name is available. |
| 186 | """ |
| 187 | console.print( |
| 188 | "No UUID provided, project pyproject.toml not found or with error.", |
| 189 | style="bold red", |
| 190 | ) |
| 191 | |
| 192 | def _display_deployment_info(self, json_response: dict[str, Any]) -> None: |
| 193 | """ |
| 194 | Display deployment information. |
| 195 | |
| 196 | Args: |
| 197 | json_response (Dict[str, Any]): The deployment information to display. |
| 198 | """ |
| 199 | console.print("Deploying the crew...\n", style="bold blue") |
| 200 | for key, value in json_response.items(): |
| 201 | console.print(f"{key.title()}: [green]{value}[/green]") |
| 202 | console.print("\nTo check the status of the deployment, run:") |
| 203 | console.print("crewai deploy status") |
| 204 | console.print(" or") |
| 205 | console.print(f'crewai deploy status --uuid "{json_response["uuid"]}"') |
| 206 | self._open_deployment_page(json_response) |
| 207 | |
| 208 | def _display_logs(self, log_messages: list[dict[str, Any]]) -> None: |
| 209 | """ |
| 210 | Display log messages. |
| 211 | |
| 212 | Args: |
| 213 | log_messages (List[Dict[str, Any]]): The log messages to display. |
| 214 | """ |
| 215 | for log_message in log_messages: |
| 216 | console.print( |
| 217 | f"{log_message['timestamp']} - {log_message['level']}: {log_message['message']}" |
| 218 | ) |
| 219 | |
| 220 | def _open_deployment_page(self, json_response: dict[str, Any]) -> None: |
| 221 | """Open the deployment show page in the user's browser when possible.""" |
| 222 | base_url = str( |
| 223 | getattr(self.plus_api_client, "base_url", None) |
| 224 | or DEFAULT_CREWAI_ENTERPRISE_URL |
| 225 | ) |
| 226 | deployment_url = self._deployment_page_url(json_response, base_url) |
no outgoing calls
searching dependent graphs…