waft.actions.agent_actions.agent_actions module

Python API client for WAFT agent operations.

class waft.actions.agent_actions.agent_actions.AgentActions(system_name: str, verify_ssl: bool = True)

Bases: object

Programmatic interface for WAFT agent operations.

This class provides a Python API for managing WAFT agents without going through the CLI. The client connection is lazily initialized - you can create an instance and deploy the agent before attempting operations that require a running agent.

Parameters:

system_name – Name of the system to manage (must exist in waft_config.json)

property agent_api_token: str | None

Get agent API token from file if it exists.

Returns:

Token string if file exists and contains non-empty content, None otherwise.

Note

Returns None silently if file doesn’t exist or is empty. This is intentional to allow optional token configuration. Check for None before use.

static check_systems_status(system_names: list[str]) → dict[str, dict[str, Any]]

Check status of WAFT agents on multiple systems.

Parameters:

system_names – List of system names to check

Returns:

Dictionary mapping system names to their status information

create_metrics_schedule(name: str, interval: str, delivery_url: str | None = None, wshim_token: str | None = None, auth_token: str | None = None) → dict[str, Any]

Create a new schedule for metrics collection.

Parameters:

  • name – Schedule name

  • interval – Collection interval (e.g., “30s”, “1m”, “5m”)

  • delivery_url – URL to send metrics to (uses default from settings if not provided)

  • wshim_token – WShim token for authentication (reads from config if not provided)

  • auth_token – Bearer token for delivery endpoint (reads from config if not provided)

Returns:

Created schedule information

delete_metrics_schedule(schedule_id: str) → None

Delete a schedule.

Parameters:

schedule_id – Schedule ID to delete

deploy_agent(check_mode: bool = False, verbose: bool = False) → bool

Deploy WAFT agent to the system.

Parameters:

  • check_mode – Run in check mode (dry run) without making changes

  • verbose – Enable verbose output during deployment

Returns:

True if deployment succeeded, False otherwise

static deploy_to_systems(system_names: list[str], check_mode: bool = False, verbose: bool = False, verify_version: bool = True) → dict[str, Any]

Deploy WAFT agent to multiple systems.

Note

Deployments are performed sequentially, one system at a time. For large deployments, this may take considerable time. Future enhancement: Consider parallel deployment for improved performance.

Parameters:

  • system_names – List of system names to deploy to

  • check_mode – Run in check mode (dry run) without making changes

  • verbose – Enable verbose output during deployment

  • verify_version – Verify deployed version matches expected version

Returns:

  • ‘successful’: List of systems that deployed successfully

  • ’failed’: List of systems that failed deployment

  • ’version_mismatches’: List of (system, deployed_version, expected_version) tuples

  • ’status_info’: Dictionary of system status information

Return type:

Dictionary with deployment results

ensure_deployed(verbose: bool = False) → AgentActions

Ensure agent is deployed and running at the expected version.

Checks if the agent is running at the correct version. If not running or version mismatches, deploys the agent and verifies deployment.

Parameters:

verbose – Enable verbose output during deployment

Returns:

Self for method chaining

Raises:

RuntimeError – If deployment fails or version mismatch after deployment

get_agent_status() → dict[str, Any]

Get agent status and health information.

Returns:

Agent status information including version, uptime, etc.

static get_agent_version() → str

Read waft-agent version from its pyproject.toml.

get_logs(lines: int | None = 100) → dict[str, Any]

Retrieve agent logs.

Parameters:

lines – Number of log lines to retrieve. If None, returns all logs.

Returns:

Dictionary with ‘logs’, ‘lines_returned’, and ‘platform’ keys.

get_metrics_schedule(schedule_id: str) → dict[str, Any]

Get details of a specific schedule.

Parameters:

schedule_id – Schedule ID

Returns:

Schedule information

list_metrics_schedules() → list[dict[str, Any]]

List all schedules for metrics collection.

Returns:

List of schedule information dictionaries

property metrics_delivery_token: str | None

Get metrics delivery bearer token from file if it exists.

Returns:

Token string if file exists and contains non-empty content, None otherwise.

Note

Returns None silently if file doesn’t exist or is empty. This is intentional to allow optional token configuration. Check for None before use.

property metrics_delivery_url: str

Get metrics delivery URL from settings.

Returns:

Metrics delivery URL

update_metrics_schedule(schedule_id: str, name: str | None = None, interval: str | None = None, delivery_url: str | None = None, wshim_token: str | None = None, auth_token: str | None = None, enabled: bool | None = None) → dict[str, Any]

Update an existing schedule.

Parameters:

  • schedule_id – Schedule ID to update

  • name – New schedule name (optional)

  • interval – New collection interval (optional)

  • delivery_url – New delivery URL (optional)

  • wshim_token – New WShim token (optional)

  • auth_token – New bearer token (optional)

  • enabled – Enable/disable schedule (optional)

Returns:

Updated schedule information

property wshim_token: str | None

Get WShim token from file if it exists.

Returns:

Token string if file exists and contains non-empty content, None otherwise.

Note

Returns None silently if file doesn’t exist or is empty. This is intentional to allow optional token configuration. Check for None before use or generate a token using: waft agent generate-wshim <service> <team> –save