---
question: Can I scaffold a pipeline in a remote module?
kind: task
via_web: na
via_cli: available
via_python: available
via_api: na
via_config: na
ref_cli: abi new pipeline
ref_python: Pipeline
prereq: You need an in-engine module listed under `modules:` in `config.yaml`, with `TripleStoreService` in its dependencies. `abi new pipeline` refuses to run inside a remote module.
next:
  - /capabilities/build/pipelines/run-a-pipeline
  - /capabilities/build/pipelines/schedule-a-pipeline
  - /capabilities/build/knowledge-graph/generate-python-classes
---

# Create a pipeline

Pipelines transform raw data from integrations into OWL/RDF triples for the knowledge graph. They are the semantic ingestion layer: a pipeline's `run()` returns an `rdflib.Graph`.

Pipelines are `naas_abi_core` code, so they belong in an [in-engine module](/user-guide/build/modules/creating-a-module#remote-or-in-engine). `abi new pipeline` refuses to run inside a remote module; there, put the logic in a plain async function, write through `self.engine.services.triple_store`, and run it from a `@job`.

## Role in the stack

```text
Integration (raw API data)  →  Pipeline (transforms to RDF)  →  Triple Store
```

A pipeline:
- Calls an integration to fetch raw data.
- Maps that data to ontology classes and properties.
- Builds an RDF graph and returns it, writing it to the triple store when it is meant to persist it.
- Offers itself to agents as tools through `as_tools()` and, once `run` is implemented, as an API endpoint.

If your output is not RDF, write a [workflow](/user-guide/build/agents/creating-a-workflow) instead.

## Web UI

Not applicable. Nexus has no screen that writes a pipeline: it is a Python class in an in-engine module. Use the CLI or Python on this page.

## CLI

### Scaffold the pipeline

From the project root:

```bash
abi new pipeline Items src/my_module/pipelines
```

It writes `ItemsPipeline.py` (the name is converted to PascalCase and `Pipeline` is appended): a configuration dataclass, a parameters model and a `Pipeline` subclass whose `run()` and `as_tools()` you fill in. The arguments are `<name> [path]`; the path defaults to the current directory. Until `run` is filled in, the pipeline is not published over REST. The command is [`abi new pipeline`](/user-guide/reference/cli#abi-new-pipeline). The full command list is the [CLI reference](/user-guide/reference/cli).


## Python

### Write the class

```python title="pipelines/ItemsPipeline.py"
from dataclasses import dataclass
from typing import Annotated

from langchain_core.tools import BaseTool, StructuredTool
from my_module import ABIModule
from my_module.integrations.MyIntegration import MyIntegration, MyIntegrationConfiguration
from naas_abi_core.pipeline import Pipeline, PipelineConfiguration, PipelineParameters
from pydantic import Field
from rdflib import RDF, RDFS, Graph, Literal, Namespace, URIRef

MY = Namespace("http://ontology.naas.ai/my_module/")  # your module's ontology namespace
GRAPH = URIRef("http://ontology.naas.ai/graph/my_module")


@dataclass
class ItemsPipelineConfiguration(PipelineConfiguration):
    """Configuration for ItemsPipeline."""


class ItemsPipelineParameters(PipelineParameters):
    filter: Annotated[
        str | None, Field(description="Optional filter for the items to ingest")
    ] = None


class ItemsPipeline(Pipeline):
    """Ingest the items of MyService into the knowledge graph."""

    __configuration: ItemsPipelineConfiguration

    def __init__(self, configuration: ItemsPipelineConfiguration):
        super().__init__(configuration)
        self.__configuration = configuration
        module = ABIModule.get_instance()
        self.__triple_store = module.engine.services.triple_store
        self.__integration = MyIntegration(
            MyIntegrationConfiguration(api_key=module.configuration.my_api_key)
        )

    def run(self, parameters: ItemsPipelineParameters) -> Graph:  # type: ignore[override]
        # 1. Fetch raw data
        items = self.__integration.list_items(filter=parameters.filter)

        # 2. Build an RDF graph
        g = Graph()
        for item in items:
            item_uri = URIRef(f"http://ontology.naas.ai/abi/item/{item['id']}")
            g.add((item_uri, RDF.type, MY.Item))  # a class of your ontology
            if item.get("name"):
                g.add((item_uri, RDFS.label, Literal(item["name"])))
            g.add((item_uri, MY.item_id, Literal(item["id"])))

        # 3. Store it, and return it
        self.__triple_store.insert(g, GRAPH)
        return g

    def as_tools(self) -> list[BaseTool]:
        return [
            StructuredTool(
                name="ingest_myservice_items",
                description="Fetch items from MyService and store them in the knowledge graph.",
                func=lambda **kwargs: len(self.run(ItemsPipelineParameters(**kwargs))),
                args_schema=ItemsPipelineParameters,
            )
        ]
```

### Let the engine load it

- **Discovery.** The engine finds every `Pipeline` subclass under `pipelines/` and instantiates it with `New()` if the class defines one, else with an empty `<Name>PipelineConfiguration()`. A pipeline whose configuration needs arguments the engine cannot supply is skipped and logged, which is why the example reads its services from the module in `__init__`.
- **REST.** `as_api` is inherited from `Expose`: when `run(parameters)` is a real override with a Pydantic parameters model, the API mounts `POST /pipelines/<class name in snake case>` (here `POST /pipelines/items_pipeline`). The returned graph comes back as `{"format": "turtle", "data": "..."}`. See the [REST API](/user-guide/build/applications/api).
- **Tools.** `as_tools()` gives an agent the pipeline as a tool: see [Creating an Agent](/user-guide/build/agents/creating-an-agent).
- **From a terminal.** `abi run script my_script.py` loads the engine and runs a script of yours, which can build the pipeline and call `run()` (in a NATS project it needs the broker).


## REST API

Not applicable. There is no HTTP call that writes the pipeline class. Once the engine loads a pipeline whose `run` takes a Pydantic model, it gets its own route, `POST /pipelines/items_pipeline`, with the same bearer key as agents: see [Run a pipeline](/user-guide/build/pipelines/run-a-pipeline?via=api). The shipped process list is the [API](/api).

## Config

Not applicable. The class is not a `config.yaml` block.

## Using onto2py generated classes

`onto2py` generates typed Python classes from your module's ontology (see [Generate Python classes](/user-guide/build/knowledge-graph/generate-python-classes)). Each class is a Pydantic model with one field per data property and an `rdf()` method that returns its triples, so a pipeline can build entities instead of raw triples:

```python
# generated from ontologies/modules/MyModuleOntology.ttl
from my_module.ontologies.modules.MyModuleOntology import Item

item = Item(_uri=f"http://ontology.naas.ai/abi/item/{item_id}", label=name, item_id=item_id)
g += item.rdf()
```

Omit `_uri` and the entity gets a generated IRI. Property names are the ontology's, in snake case.

---

## Scheduling a pipeline

A pipeline has no schedule of its own: a [module job](/user-guide/build/modules/module-jobs) on the module that owns it calls `run` on a cron or an interval, in a NATS project. The steps are on [Schedule a pipeline](/user-guide/build/pipelines/schedule-a-pipeline), and each run shows in the System app's Jobs tab ([Monitor runs](/user-guide/build/pipelines/monitor-runs)).

---

## Best practices

- **Idempotent**: running the same pipeline twice should not create duplicate triples. Use `owl:hasKey` in your ontology for entity deduplication, or check for existing triples before inserting.
- **Schema-first**: define (or reuse) an ontology class for every entity type you ingest before writing the pipeline.
- **Small batches**: for large datasets, paginate the integration and commit triples in batches to avoid memory pressure.
- **Log progress**: use `from naas_abi_core import logger`; avoid `print()`.