--- question: How do I write a workflow class in a module? kind: task via_web: na via_cli: available via_python: available via_api: na via_config: na ref_cli: abi new workflow ref_python: Workflow prereq: You need a module listed under `modules:` in `config.yaml`. See [Creating a Module](/v2/capabilities/build/modules/creating-a-module). next: - /v2/capabilities/build/workflows/expose-as-a-tool - /v2/capabilities/build/workflows/run-a-workflow - /v2/capabilities/build/workflows/write-a-durable-workflow --- > **You are reading the v2.x docs (engine modules). New projects use v3.** [Read this page in v3](/capabilities/build/agents/creating-a-workflow) · [Migrate from 2.x to 3.0](/capabilities/upgrade/migrate-to-3) # Create a workflow Workflows implement business logic. They sit at the top of the stack, orchestrating integrations and querying the knowledge graph to fulfill specific user intents. A workflow class is exposed as agent tools, as an engine API route and, on ABI 3.x in a NATS project, as the durable workflow `workflow.` that Nexus and other modules start by name. The class is `naas_abi_core.workflow.Workflow`, with a plain `run` method. It is the same on ABI 2.x and 3.x: | | ABI 2.x | ABI 3.x | |---|---|---| | `abi new workflow` writes | The class, with an `as_tools()` to fill in | The same class | | Agent tools | `as_tools()` | `as_tools()` | | HTTP | `POST /workflows/` on the engine API | `POST /workflows/` on the engine API | | Durable `workflow.` | None | In a NATS project (`nats.workflows.bridge`, on by default) | For work in several steps that must survive a crash, or that spreads over other modules, write a [durable workflow](/v2/capabilities/build/workflows/write-a-durable-workflow) (`@workflow`, ABI 3.x with NATS) instead. ## Role in the stack ```text ┌─────────────────┐ │ Workflows │ ← Business logic. Query graph, call integrations, compose results. ├─────────────────┤ │ Pipelines │ ← Data ingestion. Transform raw data into RDF. ├─────────────────┤ │ Integrations │ ← External API communication. No business logic. └─────────────────┘ ``` A workflow: - Implements a specific, named intent (e.g. "reward long-term customers"). - Can query the triple store directly with SPARQL. - Can call integrations for write operations or real-time data. - Returns a deterministic result for the same inputs. - Is exposed as a LangChain tool (for agents), as a FastAPI endpoint once `run` is implemented, and as the durable workflow `workflow.` on ABI 3.x in a NATS project. If your output is RDF, write a [pipeline](/v2/capabilities/build/knowledge-graph/creating-a-pipeline) instead. ## Web UI Not applicable. Nexus has no screen that writes a workflow: it is a Python class in a module. Use the CLI or Python on this page. On ABI 3.x in a NATS project, once the module runs, Nexus starts it from System › Jobs: see [Run a workflow](/v2/capabilities/build/workflows/run-a-workflow?via=web). ## CLI ### Scaffold the workflow From the module's folder: ```bash abi new workflow Customers workflows ``` It writes `CustomersWorkflow.py`: the name is converted to PascalCase and `Workflow` is appended. The arguments are ` [path]`. The file holds three classes: a configuration, the parameters, and the workflow. The command is [`abi new workflow`](/user-guide/reference/cli#abi-new-workflow). The path defaults to the current directory, on ABI 2.x and 3.x. The class is `naas_abi_core` code. Its `run` is a stub, and its `as_tools()` returns one tool named after the class with a placeholder description: fill both in as in the Python steps. No test file is written. --- ## Python ### Write the class ```python title="workflows/CustomersWorkflow.py" from dataclasses import dataclass from typing import Annotated from langchain_core.tools import BaseTool, StructuredTool from my_module import ABIModule from naas_abi_core.workflow import Workflow, WorkflowConfiguration from naas_abi_core.workflow.workflow import WorkflowParameters from pydantic import Field @dataclass class CustomersWorkflowConfiguration(WorkflowConfiguration): """Configuration for CustomersWorkflow.""" class CustomersWorkflowParameters(WorkflowParameters): entity_id: Annotated[ str, Field(..., description="The ID of the entity to summarize") ] class CustomersWorkflow(Workflow[CustomersWorkflowParameters]): """Summarize an entity from the knowledge graph.""" __configuration: CustomersWorkflowConfiguration def __init__(self, configuration: CustomersWorkflowConfiguration): super().__init__(configuration) self.__configuration = configuration # Services come from the module, declared in its `dependencies`. self.__triple_store = ABIModule.get_instance().engine.services.triple_store def run(self, parameters: CustomersWorkflowParameters) -> dict | list[dict]: sparql = f""" PREFIX rdfs: SELECT ?label WHERE {{ rdfs:label ?label . }} """ results = self.__triple_store.query(sparql) return {"entity_id": parameters.entity_id, "labels": [str(row[0]) for row in results]} def as_tools(self) -> list[BaseTool]: return [ StructuredTool( name="summarize_entity", description=( "Summarize an entity from the knowledge graph. " "Use when asked for information about an entity." ), func=lambda **kwargs: self.run(CustomersWorkflowParameters(**kwargs)), args_schema=CustomersWorkflowParameters, ) ] ``` ### Let the engine load it - **Discovery.** The engine finds every `Workflow` subclass under `workflows/` and instantiates it with `New()` if the class defines one, else with an empty `WorkflowConfiguration()`. A workflow whose configuration needs arguments the engine cannot supply is skipped (logged at debug level only), which is why the example reads its services from the module in `__init__` instead of taking them in the configuration. - **Tools.** An agent gets the tools with `CustomersWorkflow(CustomersWorkflowConfiguration()).as_tools()`: see [Expose as a tool](/v2/capabilities/build/workflows/expose-as-a-tool). - **REST.** `as_api` is inherited from `Expose`: when `run(parameters)` is a real override whose parameter is a Pydantic model, the API mounts `POST /workflows/` (here `POST /workflows/customers_workflow`) with the same bearer authentication as the agents. A `run` that is only `pass` or `raise NotImplementedError` is not published. See [Expose as an API](/v2/capabilities/build/workflows/expose-as-an-api). - **Durable workflow.** On ABI 3.x in a NATS project, the same live, typed `run` is also published as `workflow.customers_workflow`, which System › Jobs, `/api/workflows` and other modules start. A class whose `run` is `async` is not published there. See [Run a workflow](/v2/capabilities/build/workflows/run-a-workflow). --- ## REST API Not applicable. There is no HTTP call that writes the workflow class. Once the engine loads a workflow whose `run` takes a Pydantic model, it gets `POST /workflows/`, and on ABI 3.x in a NATS project every published class can be started through Nexus' `/api/workflows`: see [Expose as an API](/v2/capabilities/build/workflows/expose-as-an-api). ## Config Not applicable. The class is not a `config.yaml` block. On ABI 3.x, `nats.workflows.bridge: false` stops the engine from publishing its modules' classes as durable workflows. ## SPARQL as deterministic tools For read-only workflows that are just structured queries, you do not need Python. Declare the query in the module's ontology folder and the `templatablesparqlquery` module turns it into an agent tool at startup. The scaffolded module already holds an example in `ontologies/queries/`, with names prefixed by the module. In short: ```turtle intentMapping:findRecordByIdQuery a intentMapping:TemplatableSparqlQuery ; rdfs:label "find_record_by_id"@en ; intentMapping:intentDescription "Look up a single record by its identifier." ; intentMapping:sparqlTemplate """ SELECT ?record ?recordId WHERE { ... FILTER(STR(?recordId) = "{{ record_id }}") } """ ; intentMapping:hasArgument intentMapping:recordIdArg . ``` See [SPARQL Query Template System](/v2/capabilities/build/knowledge-graph/sparql_query_template_system). --- ## Composing workflows Inside one module, a workflow class can call another class's `run` directly: ```python class RewardCustomersWorkflow(Workflow[RewardCustomersParameters]): def __init__(self, configuration: RewardCustomersWorkflowConfiguration): super().__init__(configuration) self.__find_customers = FindCustomersWorkflow(FindCustomersWorkflowConfiguration()) self.__credits_workflow = AddCreditsWorkflow(AddCreditsWorkflowConfiguration()) def run(self, parameters: RewardCustomersParameters) -> dict: # Step 1: find eligible customers customers = self.__find_customers.run( FindCustomersParameters(min_months=parameters.min_months) ) rewarded = [] for customer in customers: # Step 2: add credits self.__credits_workflow.run(AddCreditsParameters( customer_id=customer["id"], credits=parameters.credits, )) rewarded.append(customer["id"]) return {"rewarded": len(rewarded), "customer_ids": rewarded} ``` A crash halfway through runs everything again. When the steps are costly, or call workflows of other modules, write a [durable workflow](/v2/capabilities/build/workflows/write-a-durable-workflow) instead (ABI 3.x, NATS project): each `ctx.step` is stored, and `ctx.call` starts another module's workflow as a child run. --- ## Running a workflow on a schedule | Line | How | |---|---| | ABI 3.x, NATS project | A `@workflow` or a `@job` on the module that calls the class's `run`. See [Schedule a workflow](/v2/capabilities/build/workflows/schedule-a-workflow). | | ABI 2.x, or 3.x without NATS | A Dagster orchestration (`abi new orchestration`) that calls `run`. See [Dagster](/v2/capabilities/build/applications/dagster). On 3.x Dagster is legacy and is removed in the next major. | --- ## Best practices - **One intent per workflow**: `GetCustomersByRevenue`, `SendOnboardingEmail`, `UpdateCompanyProfile` - each focused, each composable. - **Validate parameters**: Pydantic validation is free; use `Field(description=...)` so agents know how to call the tool correctly. - **Return structured data**: return dicts or Pydantic models. Agents and API consumers both benefit. - **Log with context**: use `from naas_abi_core import logger` and `logger.info(...)`, not `print()`. - **Test `run()` directly**: build the workflow and call `run(...)` with a parameters instance, in `Workflow_test.py` next to the file. See [Test a workflow](/v2/capabilities/build/workflows/test-a-workflow). - **Safe to repeat**: a durable run (ABI 3.x, NATS) is delivered at least once. Write to keys derived from the parameters, so a second run changes nothing.