AccountHealthWorkflow
What it is
- A non-functional template workflow for account health monitoring and intervention in the customer success manager domain.
- Provides placeholders for future workflow logic and returns a structured template response.
- Logs warnings on initialization and execution to indicate it is not implemented.
Public API
-
@dataclass AccountHealthWorkflowConfiguration(WorkflowConfiguration)- Purpose: Configuration container for
AccountHealthWorkflow. - Notes: Currently empty (
pass).
- Purpose: Configuration container for
-
class AccountHealthWorkflow(Workflow)__init__(config: Optional[AccountHealthWorkflowConfiguration] = None)- Purpose: Initialize the workflow; uses a default
AccountHealthWorkflowConfigurationif none provided. - Side effects: Logs a warning that the workflow is not functional.
- Purpose: Initialize the workflow; uses a default
async execute(inputs: Dict[str, Any]) -> Dict[str, Any]- Purpose: Template execution method; returns a dictionary describing planned steps and received input keys.
- Side effects: Logs a warning that execution is not implemented.
- Expected
inputskeys (documented):domain_specific_input,context,parameters.
get_workflow_description() -> str- Purpose: Returns a human-readable description of the workflow.
Configuration/Dependencies
- Depends on:
naas_abi_core.loggerfor warnings.naas_abi_core.workflow.workflow.WorkflowandWorkflowConfigurationbase classes.
Usage
import asyncio
from naas_abi_marketplace.domains.customer_success_manager.workflows.AccountHealthWorkflow import (
AccountHealthWorkflow,
)
async def main():
wf = AccountHealthWorkflow()
result = await wf.execute(
{
"domain_specific_input": {"account_id": "A-123"},
"context": {"region": "EMEA"},
"parameters": {"verbose": True},
}
)
print(result)
asyncio.run(main())
Caveats
- Not functional yet:
execute()does not implement account health logic; it returns a template payload with:status: "template_only"messageplanned_stepsinputs_received(list of input keys)