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

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): A dataclass defining connection fields (host, port, credentials) and implementing create_connector() to return a connector instance.
• BaseConnector (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) instead.
• ConnectorManager (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.

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.

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, 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 and add an import inside the on_init() method or at module level:

# 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.

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) 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) 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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}