kfp-server-api

import os from kfp_server_api import Configuration, ApiClient from kfp_server_api.api.pipeline_service_api import PipelineServiceApi # Replace with your KFP API endpoint. E.g., 'http://localhost:8080/pipeline' # or 'https://<your-kfp-host>/pipeline' KFP_HOST = os.environ.get('KFP_SERVER_API_HOST', 'http://localhost:8080') # Configure API key authorization: Bearer # For most KFP deployments, you might need a service account token or session cookie. # Example for no auth or basic local setup: configuration = Configuration(host=KFP_HOST) # For token-based auth (e.g., from a service account or user session): # KFP_API_TOKEN = os.environ.get('KFP_SERVER_API_TOKEN') # if KFP_API_TOKEN: # configuration.access_token = KFP_API_TOKEN # Create an API client with ApiClient(configuration) as api_client: # Create an instance of the PipelineServiceApi pipeline_api = PipelineServiceApi(api_client) try: # List pipelines (requires appropriate authentication and permissions) list_response = pipeline_api.list_pipelines() print(f"Successfully connected to KFP API host: {KFP_HOST}") print(f"Found {len(list_response.pipelines or [])} pipelines.") for p in (list_response.pipelines or []): print(f" - {p.display_name} (ID: {p.pipeline_id})") except Exception as e: print(f"Error connecting to KFP API or listing pipelines: {e}") print("Make sure KFP_SERVER_API_HOST is correctly set and you have necessary authentication.")