> ## Documentation Index
> Fetch the complete documentation index at: https://docs.noxus.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Creating data sources
> Author a knowledge-base data source in a plugin — pull files from an external system into a Noxus KB
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](/developers/plugins/sandbox-execution) 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`.
```python theme={null}
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](#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
```python theme={null}
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:
```python theme={null}
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`:
```python theme={null}
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](/developers/plugins/creating-integrations).
## Registering the data source
Return your data source classes from the plugin's `datasources()` method:
```python theme={null}
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.
## Related
The sandbox model and the file callbacks `fetch` relies on.
The `File` model and the file helper in depth.