# How to Integrate Custom Data Sources with DB-GPT's Datasource Connector System

> Learn to integrate custom data sources with DB-GPT. Extend parameters and connector classes, then use auto_register_resource for seamless integration with the ConnectorManager.

- Repository: [eosphoros/DB-GPT](https://github.com/eosphoros-ai/db-gpt)
- Tags: how-to-guide
- Published: 2026-02-23

---

**To integrate custom data sources with DB-GPT, implement a parameters dataclass extending `BaseDatasourceParameters`, create a connector class inheriting from `BaseConnector`, and register it using the `@auto_register_resource` decorator so the `ConnectorManager` can discover and instantiate it automatically.**

DB-GPT abstracts every external data store behind a unified **connector** interface, enabling plug-and-play support for proprietary databases, internal APIs, or specialized storage systems. When you integrate custom data sources with DB-GPT, you extend the base classes defined in `dbgpt-core` and leverage the AWEL auto-registration system to expose your connector to both the web UI and programmatic APIs without modifying frontend code.

## DB-GPT Connector Architecture Overview

The datasource system relies on three core components working together in `packages/dbgpt-core/src/dbgpt/datasource/`:

- **`BaseDatasourceParameters`** ([`parameter.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/parameter.py)): A dataclass defining connection fields (host, port, credentials) and implementing `create_connector()` to return a connector instance.
- **`BaseConnector`** ([`base.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/base.py)): The abstract base class defining the contract for executing queries via methods like `run()` and `run_to_df()`. For relational databases, extend **`RDBMSConnector`** ([`rdbms/base.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/rdbms/base.py)) instead.
- **`ConnectorManager`** ([`packages/dbgpt-serve/src/dbgpt_serve/datasource/manages/connector_manager.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-serve/src/dbgpt_serve/datasource/manages/connector_manager.py)): The central registry that walks the subclass tree of `BaseConnector` on startup, maps `db_type` strings to classes, and handles instantiation via `create_connector()`.

## Step 1: Define Connection Parameters

Create a dataclass inheriting from `BaseDatasourceParameters` to define your connection schema and instantiate the connector. Store this in a module like [`packages/dbgpt-ext/src/dbgpt_ext/datasource/conn_mynosql.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-ext/src/dbgpt_ext/datasource/conn_mynosql.py).

```python
from dataclasses import dataclass, field
from dbgpt.datasource.parameter import BaseDatasourceParameters
from dbgpt.core.awel.flow import auto_register_resource, ResourceCategory, TAGS_ORDER_HIGH
from dbgpt.util.i18n_utils import _

@auto_register_resource(
    label=_("MyNoSQL datasource"),
    category=ResourceCategory.DATABASE,
    tags={"order": TAGS_ORDER_HIGH},
    description=_("Connector for a custom NoSQL store."),
)
@dataclass
class MyNoSQLParameters(BaseDatasourceParameters):
    """Connection parameters for MyNoSQL."""

    __type__ = "mynosql"

    endpoint: str = field(metadata={"help": _("API endpoint, e.g. https://api.example.com")})
    api_key: str = field(
        metadata={"help": _("API key, can be an env var like ${env:MYNOSQL_API_KEY}")},
    )

    def create_connector(self) -> "MyNoSQLConnector":
        return MyNoSQLConnector.from_parameters(self)

```

The `__type__` attribute serves as the unique identifier DB-GPT uses when routing datasource creation requests. The `@auto_register_resource` decorator generates `ResourceMetadata` that powers the UI's "Add Data Source" form automatically.

## Step 2: Implement the Connector Class

Implement a class extending `BaseConnector` (or `RDBMSConnector` for SQL databases) in the same file. You must define `db_type`, `driver`, and `param_class()`, plus the execution methods.

```python
from typing import Any, List
import json
import requests
from dbgpt.datasource.base import BaseConnector

class MyNoSQLConnector(BaseConnector):
    """HTTP-based NoSQL connector implementing the BaseConnector contract."""

    db_type = "mynosql"
    driver = "http"

    @classmethod
    def param_class(cls):
        return MyNoSQLParameters

    def __init__(self, endpoint: str, api_key: str):
        self.endpoint = endpoint
        self.api_key = api_key

    @classmethod
    def from_parameters(cls, parameters: MyNoSQLParameters):
        return cls(parameters.endpoint, parameters.api_key)

    def run(self, command: str, fetch: str = "all") -> List:
        """Execute a query against the custom data source.
        
        Args:
            command: JSON string representing the query payload.
            fetch: Ignored in this implementation, kept for API compatibility.
        """
        payload = json.loads(command)
        resp = requests.post(
            self.endpoint, 
            json=payload,
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        resp.raise_for_status()
        return resp.json()

```

The `run()` method is the minimal required interface. For SQL-based connectors, you would also implement `run_to_df()` to return **Pandas DataFrames** directly.

## Step 3: Register with AWEL

The `@auto_register_resource` decorator applied in Step 1 handles registration automatically when Python imports the class. This decorator, defined in [`packages/dbgpt-core/src/dbgpt/core/awel/flow/base.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-core/src/dbgpt/core/awel/flow/base.py), attaches metadata to your class including display labels, categories, and descriptions that the UI consumes.

No additional registration logic is required in your connector file.

## Step 4: Enable Discovery in ConnectorManager

To ensure DB-GPT discovers your connector on startup, you must make the class importable by the `ConnectorManager`. Modify [`packages/dbgpt-serve/src/dbgpt_serve/datasource/manages/connector_manager.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-serve/src/dbgpt_serve/datasource/manages/connector_manager.py) and add an import inside the `on_init()` method or at module level:

```python

# Inside connector_manager.py

def on_init(self):
    # Existing imports...

    from dbgpt_ext.datasource.conn_mynosql import MyNoSQLConnector  # noqa: F401

```

The `ConnectorManager.on_init()` method walks the subclass tree of `BaseConnector` (lines 44-80 in the source) to build the internal `_supported_types` mapping. Importing your module triggers the decorator and populates this registry.

## Step 5: Create and Test the Datasource

Once registered, create datasource instances via the Python client or REST API. The `ConnectorManager._create_parameters()` method instantiates your parameters class, and `create_connector()` invokes your `create_connector()` implementation.

```python
import asyncio
from dbgpt_client import Client
from dbgpt_client.datasource import create_datasource, list_datasource

async def main():
    client = Client(api_key="dbgpt")
    
    # Define datasource configuration matching __type__ and field names

    ds_config = {
        "name": "production_nosql",
        "type": "mynosql",
        "params": {
            "endpoint": "https://api.example.com/query",
            "api_key": "${env:MYNOSQL_API_KEY}"
        }
    }
    
    # Test connectivity before persisting

    await client.test_datasource(ds_config)
    
    # Save to DB-GPT's internal config database

    await create_datasource(client, ds_config)
    
    # Verify registration

    all_ds = await list_datasource(client)
    print(f"Available datasources: {[d.name for d in all_ds]}")

asyncio.run(main())

```

The `test_datasource()` call executes `ConnectorManager.test_connection()`, which validates credentials by invoking your connector's `run()` method with a lightweight probe query.

## Summary

- **Extend `BaseDatasourceParameters`** to define connection fields and instantiate your connector via `create_connector()`.
- **Subclass `BaseConnector`** (or `RDBMSConnector` for SQL stores) and implement `db_type`, `param_class()`, and execution methods like `run()`.
- **Apply `@auto_register_resource`** to expose metadata to the UI automatically without frontend changes.
- **Import your module** in `ConnectorManager.on_init()` to register the class in the global type map.
- **Use the standard client API** to create, test, and query datasources using your custom connector.

## Frequently Asked Questions

### Do I need to modify DB-GPT's frontend code to add a custom connector?

No. The `@auto_register_resource` decorator generates `ResourceMetadata` that the frontend consumes automatically. When you add a connector, the web UI's "Add Data Source" flow lists your connector using the `label` and `description` from the decorator, rendering forms based on your dataclass fields without requiring any JavaScript or React modifications.

### What is the difference between `BaseConnector` and `RDBMSConnector`?

**`BaseConnector`** ([`packages/dbgpt-core/src/dbgpt/datasource/base.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-core/src/dbgpt/datasource/base.py)) provides the minimal interface for any data store, requiring only `run()` and basic lifecycle methods. **`RDBMSConnector`** ([`packages/dbgpt-core/src/dbgpt/datasource/rdbms/base.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-core/src/dbgpt/datasource/rdbms/base.py)) extends this with SQL-specific abstractions like `run_to_df()`, schema introspection methods (`get_table_info()`, `get_indexes()`), and dialect handling. Use `RDBMSConnector` for relational databases and `BaseConnector` for NoSQL, REST APIs, or custom binary protocols.

### How does DB-GPT handle sensitive fields like passwords or API keys?

Mark sensitive fields in your parameters dataclass with appropriate metadata. The base parameter system in [`packages/dbgpt-core/src/dbgpt/datasource/parameter.py`](https://github.com/eosphoros-ai/DB-GPT/blob/main/packages/dbgpt-core/src/dbgpt/datasource/parameter.py) automatically tags fields containing secret patterns with `privacy` flags. The UI hides these inputs as password fields, and the system supports environment variable substitution (e.g., `${env:API_KEY}`) to keep secrets out of configuration databases.

### Can I use a custom connector inside an AWEL flow?

Yes. Inject the `ConnectorManager` into your flow operators and retrieve live connector instances by name using `mgr.get_connector("datasource_name")`. This returns your initialized connector class (e.g., `MyNoSQLConnector`), allowing you to call `run()` directly within `MapOperator` or `JoinOperator` nodes. The connector instance is thread-safe and manages its own connection pooling according to your implementation.