Creating data sources

A data source is an external system a Knowledge Base ingests files from. A plugin data source lets you feed documents from a proprietary system, internal store, or niche SaaS into the KB pipeline. The platform owns the KB itself — chunking, embedding, retrieval — and your data source only answers one question: which files should be added?

A data source’s fetch() runs inside the plugin’s sandbox worker, invoked over JSON-RPC (datasource.fetch). Read How plugins run for the execution model — network posture, logging, and especially the file callbacks that move bytes across the sandbox boundary.

The interface

A data source subclasses BaseDataSource[ConfigType] and implements fetch.

from noxus_sdk.datasources import BaseDataSource, DatasourceConfiguration
from noxus_sdk.ncl import Parameter, ConfigText
from noxus_sdk.files import File
from noxus_sdk.plugins.context import RemoteExecutionContext


class MyDataSourceConfig(DatasourceConfiguration):
    folder: str = Parameter(default="root", display=ConfigText(label="Folder"))


class MyDataSource(BaseDataSource[MyDataSourceConfig]):
    datasource_name = "MyDataSource"        # unique id
    title = "My Data Source"
    description = "Ingest files from My System"

    async def fetch(self, ctx: RemoteExecutionContext) -> list[File]:
        helper = ctx.get_file_helper()
        files: list[File] = []
        for name, content in _pull_from_my_system(self.config.folder):
            f = await helper.upload_file(
                file_name=name,
                content=content,               # bytes
                content_type="text/plain",
                group_id=ctx.group_id,
            )
            files.append(f)
        return files

Class attributes

Attribute	Meaning
`datasource_name`	Unique identifier for the data source type (used for lookup and in the manifest).
`title` / `description`	Shown in the KB “Add knowledge” UI.
`image`	Optional icon URL.
`integrations`	Credential types this data source needs (read via `ctx.get_integration_credentials(...)`).
`supports_sync`	Reserved. Leave `False` (the default) — see Sync model.

ConfigType is your DatasourceConfiguration subclass (a NodeConfiguration, so fields are Parameter(...) with optional display= widgets). It reaches your instance as self.config.

`fetch` contract

async def fetch(self, ctx: RemoteExecutionContext) -> list[File]:

fetch performs a one-shot ingestion: pull the files you want in the KB and return them as File descriptors. This is the “Add knowledge” flow — the user picks your data source, fills its config, and the platform ingests whatever fetch returns. The important rule: you don’t return bytes, you upload them. Fetch each file’s content, then persist it with the file helper:

f = await ctx.get_file_helper().upload_file(
    file_name="report.pdf",
    content=pdf_bytes,
    content_type="application/pdf",
    group_id=ctx.group_id,
)

upload_file stores the bytes on the platform (over a host callback — the sandbox has no direct storage access) and returns a File descriptor. Return the list of descriptors; the platform then chunks and embeds them into the KB. The upload is workspace-scoped by the host, so pass ctx.group_id for the calling workspace. Raise from fetch to fail the ingestion with a user-visible message.

Using credentials

Most data sources talk to an authenticated system. Declare the credential type and read it in fetch:

class MyDataSource(BaseDataSource[MyDataSourceConfig]):
    integrations = ["my_system"]

    async def fetch(self, ctx):
        creds = ctx.get_integration_credentials("my_system")
        client = MyClient(token=creds.get("token"))
        ...

The credential type comes from a BaseIntegration / BaseCredentials pair in the same plugin — see Creating integrations.

Registering the data source

Return your data source classes from the plugin’s datasources() method:

class MyPlugin(BasePlugin[MyPluginConfig]):
    ...
    def datasources(self) -> list[type[BaseDataSource]]:
        return [MyDataSource]

A plugin that provides a data source satisfies the “at least one node, integration, or data source” requirement on its own. Each data source is serialized into the manifest (DatasourceDefinition); regenerate manifest.json whenever its name or config changes. On the platform side a single generic Plugin Datasource node backs every plugin-provided data source and slots into the existing KB ingestion path — there’s nothing extra to wire; the platform resolves your data source by name from the manifest and dispatches datasource.fetch to the worker.

Sync model

Only one-shot fetch is supported today. Incremental sync — where the platform’s sync engine periodically polls the source for changes and adds/updates/removes documents (a list/get/download interface) — is a later phase. supports_sync is the reserved flag for it; leave it False. Until then, re-running “Add knowledge” is how content is refreshed.

Testing

Unit-test fetch directly with a stubbed file helper, and exercise the full ingestion end to end by installing the plugin against a running stack (the reference plugin under tests/plugins/ includes a data source and its run_plugin_e2e.py driver ingests from it). noxus plugin serve also mirrors datasource.fetch for local development.

How plugins run

The sandbox model and the file callbacks fetch relies on.

Working with files

The File model and the file helper in depth.

Overview

Tutorial

Reference

GitOps

The interface

Class attributes

`fetch` contract

Using credentials

Registering the data source

Sync model

Testing

How plugins run

Working with files

​The interface

​Class attributes

​fetch contract

​Using credentials

​Registering the data source

​Sync model

​Testing

​Related

How plugins run

Working with files

The interface

Class attributes

`fetch` contract

Using credentials

Registering the data source

Sync model

Testing

Related