Argo Workflows Python Client
The `argo-workflows` Python client provides programmatic access to the Argo Workflows API, enabling users to create, manage, and monitor workflows and other resources. It is an OpenAPI-generated client, meaning its version (currently 6.6.19) is closely tied to the Argo Workflows server's API specification. The library is actively maintained, with frequent updates to keep pace with new server features and API changes.
Common errors
-
argo_workflows.ApiException: (401) Reason: Unauthorized
cause The API client failed to authenticate with the Argo Workflows server.fixCheck that the `ARGO_WORKFLOWS_TOKEN` environment variable is set and contains a valid bearer token for the Argo Workflows API. Ensure the token has the necessary permissions. If using other authentication methods, verify those credentials. -
urllib3.exceptions.MaxRetryError: HTTPSConnectionPool(...) Read timed out. (Caused by SSLError(CertificateError("hostname '...' doesn't match '...'",)))cause SSL certificate verification failed, often due to a mismatch between the hostname in the certificate and the host URL or an untrusted certificate.fixSet `verify_ssl=True` in your `Configuration` and provide the path to your CA certificate via `ssl_ca_cert`. Alternatively, if you understand the security risks and for development purposes, set `verify_ssl=False`. -
AttributeError: module 'argo_workflows.model.io_argoproj_workflow_v1alpha1_workflow_spec' has no attribute 'entrypoint'
cause The client version used has a different API specification (e.g., a newer or older Argo Workflows server version) than expected, or there's a typo in the attribute name.fixVerify that your `argo-workflows` client version is compatible with your Argo Workflows server version. Update or downgrade the client as necessary. Double-check the attribute name against the official Argo Workflows API documentation. -
TypeError: 'NoneType' object is not subscriptable
cause Attempting to access an item (e.g., `workflow.metadata['name']`) on an object that is `None`, often because a required nested object was not initialized or returned `None` from an API call.fixEnsure all required nested objects (e.g., `metadata`, `spec`, `templates`) are properly initialized when constructing workflow requests. For API responses, add checks for `None` before accessing attributes or dict keys.
Warnings
- gotcha The `argo-workflows` Python client is auto-generated and its version needs to be compatible with the Argo Workflows server it connects to. Significant version mismatches can lead to API errors or `AttributeError` for missing/changed fields in models.
- gotcha Incorrect host URL or authentication details are common causes of connection failures. SSL certificate verification can also be an issue in non-standard environments.
- gotcha When constructing workflow objects, it's easy to miss required fields or provide incorrect types, leading to validation errors (either client-side or server-side). The client-side validation is enabled by default.
Install
-
pip install argo-workflows
Imports
- Configuration
from argo_workflows import Configuration
- ApiClient
from argo_workflows import ApiClient
- WorkflowServiceApi
from argo_workflows.api import workflow_service_api
- IoArgoprojWorkflowV1alpha1Workflow
from argo_workflows.model import io_argoproj_workflow_v1alpha1_workflow
- ApiException
from argo_workflows import ApiException
Quickstart
import argo_workflows
from argo_workflows.api import workflow_service_api
import os
# Configure API client
configuration = argo_workflows.Configuration(
host=os.environ.get("ARGO_WORKFLOWS_HOST", "http://localhost:2746"),
verify_ssl=os.environ.get("ARGO_WORKFLOWS_VERIFY_SSL", "false").lower() == "true",
)
# Bearer Token Authentication (common for Argo Workflows)
token = os.environ.get("ARGO_WORKFLOWS_TOKEN")
if token:
configuration.api_key["Authorization"] = token
configuration.api_key_prefix["Authorization"] = "Bearer"
else:
print("Warning: ARGO_WORKFLOWS_TOKEN not set. API calls might fail without authentication.")
# Enter a context with an instance of the API client
with argo_workflows.ApiClient(configuration) as api_client:
# Create an instance of the API class
api_instance = workflow_service_api.WorkflowServiceApi(api_client)
namespace = os.environ.get("ARGO_WORKFLOWS_NAMESPACE", "argo") # Example namespace
try:
# Example: List workflows in a given namespace
api_response = api_instance.list_workflows(namespace=namespace)
print(f"Successfully listed {len(api_response.items)} workflows in namespace '{namespace}'.")
if api_response.items:
print(f"First workflow: {api_response.items[0].metadata.name}")
except argo_workflows.ApiException as e:
print(f"Exception when calling WorkflowServiceApi->list_workflows: {e}\n")