> For the complete documentation index, see [llms.txt](https://docs.anthriq.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.anthriq.com/bxi-studio/developer/overview/backend-services/registries.md).

# Manage operators and pipelines

Two registry services back the executor. The operator registry holds installable node plugins; the pipeline registry holds reusable pipeline definitions. Both clients share a shape, with one asymmetry: operators are keyed by `name`, pipelines by `id`.

## Manage operators

| Method      | Payload                                               | Response data                            |
| ----------- | ----------------------------------------------------- | ---------------------------------------- |
| `install`   | `{ name, version, platform?, force? }`                | `{ name, version, installPath, status }` |
| `uninstall` | `{ name, version }`                                   | `{ name, version, removed }`             |
| `list`      | `{ remote?, page?, pageSize?, operator?, versions? }` | `{ operators, total, page, pageSize }`   |
| `info`      | `{ name, version, remote? }`                          | Operator metadata                        |
| `status`    | `{ name, version }`                                   | `{ name, version, status }`              |
| `push`      | `{ name, version, tarPath?, localOnly? }`             | `{ name, version, pushed, installed }`   |
| `repair`    | `{}`                                                  | `{ repaired, issues }`                   |

An operator is in one of four install states: `INSTALLED`, `NOT_INSTALLED`, `BROKEN`, or `CACHED`.

{% tabs %}
{% tab title="Node.js" %}

```typescript
import { OperatorRegistryClient } from "@anthriq_dev/services";

const registry = new OperatorRegistryClient({
  endpoint: "ipc:///tmp/bxi-operator-registry.sock",
});
await registry.connect();

await registry.install({ name: "fft", version: "1.0.0" });

const local = await registry.list();
for (const operator of local.data?.operators ?? []) {
  console.log(`${operator.name}@${operator.version} -> ${operator.installPath}`);
}

const state = await registry.status({ name: "fft", version: "1.0.0" });
console.log(state.data?.status);
```

{% endtab %}

{% tab title="Python" %}

```python
from anthriq_services import OperatorRegistryClient

async with OperatorRegistryClient("ipc:///tmp/bxi-operator-registry.sock") as registry:
    await registry.install(name="fft", version="1.0.0")

    local = await registry.list()
    for operator in local.data.get("operators", []):
        print(f"{operator['name']}@{operator['version']} -> {operator['installPath']}")

    state = await registry.status(name="fft", version="1.0.0")
    print(state.data["status"])
```

{% endtab %}
{% endtabs %}

`platform` is auto-detected when omitted. Set `force: true` to reinstall over an existing copy. Pass `remote: true` to `list` to query the remote registry instead of the local store.

### Read operator metadata

`info()` returns the full manifest. The `parameters` and `signals` fields are the contract a pipeline author writes against.

| Field                       | Type       | Description                          |
| --------------------------- | ---------- | ------------------------------------ |
| `name`, `version`           | `string`   | Identity                             |
| `description`               | `string`   | Summary                              |
| `authors`                   | `string[]` | Authors                              |
| `createdAt`                 | `string`   | Build timestamp                      |
| `os`, `arch`                | `string`   | Target platform                      |
| `license`                   | `string`   | License identifier                   |
| `size`                      | `string`   | Package size                         |
| `tags`                      | `string[]` | Search tags                          |
| `checksum`                  | `string`   | Package checksum                     |
| `entryPoints`               | object     | Per-language entry points            |
| `executable`, `entrypoint`  | `string`   | Resolved executable and entry symbol |
| `parameters`                | array      | Configuration the node accepts       |
| `signals`                   | array      | Signals the node accepts at runtime  |
| `inputTypes`, `outputTypes` | object     | Port type contracts                  |
| `artifacts`                 | array      | Files in the package                 |

{% tabs %}
{% tab title="Node.js" %}

```typescript
const meta = await registry.info({ name: "fft", version: "1.0.0" });

for (const signal of meta.data?.signals ?? []) {
  console.log(signal.name);
}
```

{% endtab %}

{% tab title="Python" %}

```python
meta = await registry.info(name="fft", version="1.0.0")

for signal in meta.data.get("signals", []):
    print(signal["name"])
```

{% endtab %}
{% endtabs %}

Cross-check `signals` before calling `signalPipeline`. The executor reports delivery rather than acceptance, so an unrecognised signal name fails silently from the caller's side. See [Run pipelines](/bxi-studio/developer/overview/backend-services/pipelines.md).

### Publish an operator

{% tabs %}
{% tab title="Node.js" %}

```typescript
await registry.push({
  name: "my-operator",
  version: "1.0.0",
  tarPath: "/build/my-operator-1.0.0.tar.gz",
  localOnly: true,
});
```

{% endtab %}

{% tab title="Python" %}

```python
await registry.push(
    name="my-operator",
    version="1.0.0",
    tar_path="/build/my-operator-1.0.0.tar.gz",
    local_only=True,
)
```

{% endtab %}
{% endtabs %}

`localOnly: true` installs into the local store without publishing, which is what you want while iterating.

### Repair a broken store

{% tabs %}
{% tab title="Node.js" %}

```typescript
const result = await registry.repair();
console.log(result.data?.repaired, result.data?.issues);
```

{% endtab %}

{% tab title="Python" %}

```python
result = await registry.repair()
print(result.data["repaired"], result.data.get("issues"))
```

{% endtab %}
{% endtabs %}

`repair()` reconciles the registry index against the files on disk. Run it when an operator reports `BROKEN`, which happens after an interrupted install or a manual file deletion.

## Manage pipeline definitions

| Method      | Payload                                                | Response data                          |
| ----------- | ------------------------------------------------------ | -------------------------------------- |
| `install`   | `{ id, version, force? }`                              | `{ id, version, installPath, status }` |
| `uninstall` | `{ id, version }`                                      | `{ id, version, removed }`             |
| `list`      | `{ remote?, page?, pageSize? }`                        | `{ pipelines, total, page, pageSize }` |
| `info`      | `{ id, version, remote?, extend? }`                    | Pipeline metadata                      |
| `status`    | `{ id, version }`                                      | `{ id, version, status }`              |
| `push`      | `{ id, version, pipelineJson?, tarPath?, localOnly? }` | `{ id, version, pushed, installed }`   |
| `pull`      | `{ id, version, force? }`                              | `{ id, version, installPath }`         |

### Register a pipeline from JSON

{% tabs %}
{% tab title="Node.js" %}

```typescript
import { PipelineRegistryClient } from "@anthriq_dev/services";

const registry = new PipelineRegistryClient({
  endpoint: "ipc:///tmp/bxi-pipeline-registry.sock",
});
await registry.connect();

await registry.push({
  id: "eeg-basic",
  version: "1.0.0",
  pipelineJson: {
    name: "Basic EXG Pipeline",
    nodes: [
      { id: "eeg", type: "node", name: "eegstream", version: "1.0.0" },
      { id: "ws", type: "node", name: "websocket", version: "1.0.0", config: { port: 8080 } },
    ],
    pipes: [{ id: "p1", source: "eeg", destination: "ws" }],
  },
});
```

{% endtab %}

{% tab title="Python" %}

```python
from anthriq_services import PipelineRegistryClient

async with PipelineRegistryClient("ipc:///tmp/bxi-pipeline-registry.sock") as registry:
    await registry.push(
        id="exg-basic",
        version="1.0.0",
        pipeline_json={
            "name": "Basic EXG Pipeline",
            "nodes": [
                {"id": "eeg", "type": "node", "name": "eegstream", "version": "1.0.0"},
                {"id": "ws", "type": "node", "name": "websocket", "version": "1.0.0",
                 "config": {"port": 8080}},
            ],
            "pipes": [{"id": "p1", "source": "eeg", "destination": "ws"}],
        },
    )
```

{% endtab %}
{% endtabs %}

`push` accepts either `pipelineJson` for an inline definition or `tarPath` for a packaged one.

> **Note:** Stored pipeline nodes name the operator with `name` and `version`. That differs from the inline format `createPipeline` accepts, where `type` names the operator directly.

### Resolve operator details

{% tabs %}
{% tab title="Node.js" %}

```typescript
const meta = await registry.info({ id: "eeg-basic", version: "1.0.0", extend: true });

console.log(meta.data?.nodes);
console.log(meta.data?.entry, meta.data?.exit);
```

{% endtab %}

{% tab title="Python" %}

```python
meta = await registry.info(id="exg-basic", version="1.0.0", extend=True)

print(meta.data["nodes"])
print(meta.data.get("entry"), meta.data.get("exit"))
```

{% endtab %}
{% endtabs %}

`extend: true` resolves each node's operator metadata inline instead of returning bare references. `entry` and `exit` declare the graph's external ports.

### Pull from the remote registry

{% tabs %}
{% tab title="Node.js" %}

```typescript
await registry.pull({ id: "eeg-basic", version: "1.0.0", force: true });
```

{% endtab %}

{% tab title="Python" %}

```python
await registry.pull(id="exg-basic", version="1.0.0", force=True)
```

{% endtab %}
{% endtabs %}

`pull` fetches and installs in one call. `install` assumes the package is already cached locally.

## Install what a pipeline needs

A pipeline definition references operators by name and version. Both must be installed before the executor can run the graph.

{% tabs %}
{% tab title="Node.js" %}

```typescript
import { BxiServicesClient } from "@anthriq_dev/services";

const services = new BxiServicesClient();
await services.initialize();

const meta = await services.pipelineRegistry.info({
  id: "eeg-basic",
  version: "1.0.0",
  extend: true,
});

for (const node of meta.data?.nodes ?? []) {
  const state = await services.operatorRegistry.status({
    name: node.name,
    version: node.version,
  });

  if (state.data?.status !== "INSTALLED") {
    await services.operatorRegistry.install({ name: node.name, version: node.version });
  }
}

await services.executor.createPipeline({
  pipelineId: "run-1",
  config: meta.data as Record<string, unknown>,
});
```

{% endtab %}

{% tab title="Python" %}

```python
from anthriq_services import BxiServicesClient

async with BxiServicesClient() as services:
    meta = await services.pipeline_registry.info(
        id="exg-basic", version="1.0.0", extend=True
    )

    for node in meta.data.get("nodes", []):
        state = await services.operator_registry.status(
            name=node["name"], version=node["version"]
        )
        if state.data["status"] != "INSTALLED":
            await services.operator_registry.install(
                name=node["name"], version=node["version"]
            )

    await services.executor.create_pipeline(pipeline_id="run-1", config=meta.data)
```

{% endtab %}
{% endtabs %}

> **Note:** The executor reports a missing operator as a pipeline creation failure, not a registry error. Check operator status first when `createPipeline` fails on a definition that used to work.

## Next steps

* [Run pipelines](/bxi-studio/developer/overview/backend-services/pipelines.md)
* [Record and play back](/bxi-studio/developer/overview/backend-services/recording.md)