Argo Workflows Python Client

6.6.19 · active · verified Thu Apr 16

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

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to configure the `argo-workflows` client for authentication (using a bearer token from an environment variable), instantiate the `WorkflowServiceApi`, and list workflows in a specified namespace. Ensure `ARGO_WORKFLOWS_HOST` and `ARGO_WORKFLOWS_TOKEN` (if using auth) are set.

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")

view raw JSON →