mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-10 16:48:30 +00:00
f5246039e5
* feat: set basic structure deploy commands * feat: add first iteration of CLI Deploy * feat: some minor refactor * feat: Add api, Deploy command and update cli * feat: Remove test token * feat: add auth0 lib, update cli and improve code * feat: update code and decouple auth * fix: parts of the code * feat: Add token manager to encrypt access token and get and save tokens * feat: add audience to costants * feat: add subsystem saving credentials and remove comment of type hinting * feat: add get crew version to send on header of request * feat: add docstrings * feat: add tests for authentication module * feat: add tests for utils * feat: add unit tests for cl * feat: add tests * feat: add deploy man tests * feat: fix type checking issue * feat: rename tests to pass ci * feat: fix pr issues * feat: fix get crewai versoin * fix: add timeout for tests.yml
64 lines
2.1 KiB
Python
64 lines
2.1 KiB
Python
from os import getenv
|
|
|
|
import requests
|
|
|
|
|
|
class CrewAPI:
|
|
"""
|
|
CrewAPI class to interact with the crewAI+ API.
|
|
"""
|
|
|
|
def __init__(self, api_key: str) -> None:
|
|
self.api_key = api_key
|
|
self.headers = {
|
|
"Authorization": f"Bearer {api_key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
self.base_url = getenv(
|
|
"CREWAI_BASE_URL", "https://dev.crewai.com/crewai_plus/api/v1/crews"
|
|
)
|
|
|
|
def _make_request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
|
|
url = f"{self.base_url}/{endpoint}"
|
|
return requests.request(method, url, headers=self.headers, **kwargs)
|
|
|
|
# Deploy
|
|
def deploy_by_name(self, project_name: str) -> requests.Response:
|
|
return self._make_request("POST", f"by-name/{project_name}/deploy")
|
|
|
|
def deploy_by_uuid(self, uuid: str) -> requests.Response:
|
|
return self._make_request("POST", f"{uuid}/deploy")
|
|
|
|
# Status
|
|
def status_by_name(self, project_name: str) -> requests.Response:
|
|
return self._make_request("GET", f"by-name/{project_name}/status")
|
|
|
|
def status_by_uuid(self, uuid: str) -> requests.Response:
|
|
return self._make_request("GET", f"{uuid}/status")
|
|
|
|
# Logs
|
|
def logs_by_name(
|
|
self, project_name: str, log_type: str = "deployment"
|
|
) -> requests.Response:
|
|
return self._make_request("GET", f"by-name/{project_name}/logs/{log_type}")
|
|
|
|
def logs_by_uuid(
|
|
self, uuid: str, log_type: str = "deployment"
|
|
) -> requests.Response:
|
|
return self._make_request("GET", f"{uuid}/logs/{log_type}")
|
|
|
|
# Delete
|
|
def delete_by_name(self, project_name: str) -> requests.Response:
|
|
return self._make_request("DELETE", f"by-name/{project_name}")
|
|
|
|
def delete_by_uuid(self, uuid: str) -> requests.Response:
|
|
return self._make_request("DELETE", f"{uuid}")
|
|
|
|
# List
|
|
def list_crews(self) -> requests.Response:
|
|
return self._make_request("GET", "")
|
|
|
|
# Create
|
|
def create_crew(self, payload) -> requests.Response:
|
|
return self._make_request("POST", "", json=payload)
|