Implementation:Mage ai Mage ai Source Connector Directory Pattern

Knowledge Sources	Mage Integrations
Domains	Data_Integration, Software_Architecture
Last Updated	2026-02-09 00:00 GMT

Overview

Pattern documentation for the standardized directory structure and file layout used by all source connectors in the Mage integrations framework.

Description

This is a Pattern Doc (not an API Doc). It documents the file structure convention that all source connectors must follow, with reference implementations from the simplest (Titanic: 34 lines) to the most complex (Chargebee: multi-file with streams).

Usage

Follow this pattern when scaffolding a new source connector directory.

Code Reference

Source Location

• Repository: mage-ai
• File: Pattern derived from existing connectors in mage_integrations/mage_integrations/sources/

Reference Implementations

Simplest: Titanic (34 lines)

# sources/titanic/__init__.py
from mage_integrations.sources.base import Source, main

class Titanic(Source):
    def load_data(self, stream, bookmarks=None, query=None, **kwargs):
        # Load from static CSV
        import pandas as pd
        df = pd.read_csv("https://raw.githubusercontent.com/.../titanic.csv")
        yield df.to_dict("records")

if __name__ == "__main__":
    main(Titanic)

SQL: PostgreSQL

# sources/postgresql/__init__.py
from mage_integrations.sources.sql.base import Source as SQLSource

class PostgreSQL(SQLSource):
    def build_connection(self):
        from mage_integrations.connections.postgresql import PostgreSQL as Connection
        return Connection(**self.config)

    def build_discover_query(self, schema="public", streams=None):
        return super().build_discover_query(schema, streams)

if __name__ == "__main__":
    from mage_integrations.sources.base import main
    main(PostgreSQL)

API: Chargebee (multi-file)

sources/chargebee/
    __init__.py        # Chargebee(Source) - overrides load_data, discover
    client.py          # ChargebeeClient(Client) - Basic Auth, rate limiting
    schemas/           # 48 JSON Schema files
    streams/
        __init__.py
        base.py        # BaseChargebeeStream - pagination, transform
        customers.py   # Customer-specific logic
        events.py      # Event-specific logic

Import

# Base classes to extend:
from mage_integrations.sources.base import Source, main
from mage_integrations.sources.sql.base import Source as SQLSource
from mage_integrations.sources.http.client import Client

I/O Contract

Inputs

Name	Type	Required	Description
Connector type decision	Choice	Yes	SQL, API, or file-based source

Outputs

Name	Type	Description
Directory structure	Files	__init__.py, schemas/, templates/, optionally client.py, streams/

Usage Examples

# Create a new SQL source connector
mkdir -p mage_integrations/mage_integrations/sources/my_database
touch mage_integrations/mage_integrations/sources/my_database/__init__.py
mkdir -p mage_integrations/mage_integrations/sources/my_database/templates
echo '{"host": "", "port": 5432, "database": "", "username": "", "password": ""}' \
    > mage_integrations/mage_integrations/sources/my_database/templates/config.json

Related Pages

Implements Principle

• Principle:Mage_ai_Mage_ai_Source_Connector_Scaffolding

Requires Environment

• Environment:Mage_ai_Mage_ai_Python_3_9_Runtime