Data Source Connectors

Relevant source files

• api/apps/restful_apis/connector_api.py
• api/db/services/connector_service.py
• common/constants.py
• common/data_source/__init__.py
• common/data_source/blob_connector.py
• common/data_source/config.py
• common/data_source/confluence_connector.py
• common/data_source/discord_connector.py
• common/data_source/dropbox_connector.py
• common/data_source/jira/connector.py
• common/data_source/models.py
• common/data_source/notion_connector.py
• common/data_source/rdbms_connector.py
• common/data_source/slack_connector.py
• common/data_source/utils.py
• common/http_client.py
• deepdoc/parser/txt_parser.py
• internal/agent/audio/model_provider_synthesizer.go
• internal/dao/connector.go
• internal/engine/redis/redis.go
• internal/engine/redis/redis_test.go
• internal/handler/connector.go
• internal/handler/connector_test.go
• internal/router/router_test.go
• internal/service/connector.go
• internal/service/connector_box_oauth_test.go
• internal/service/nlp/synonym.go
• internal/service/nlp/synonym_test.go
• internal/service/oauth_login.go
• rag/svr/sync_data_source.py
• rag/utils/opendal_conn.py
• test/testcases/restful_api/test_connector_routes_unit.py
• test/testcases/test_web_api/test_connector_app/test_connector_routes_unit.py
• test/unit_test/common/test_dropbox_connector.py
• test/unit_test/data_source/test_slack_connector_unit.py
• test/unit_test/rag/test_sync_data_source.py
• web/src/assets/svg/data-source/slack.svg
• web/src/pages/user-setting/data-source/constant/index.tsx

RAGFlow provides a robust external data source connector system that allows users to synchronize content from over 30 external platforms directly into their knowledge bases. This system automates the ingestion of documents, spreadsheets, code repositories, and communication logs, integrating them into the RAGFlow document processing pipeline.

Overview

The connector system is built on a multi-tier architecture that handles authentication, periodic polling, incremental synchronization, and error recovery. It abstracts various APIs (REST, GraphQL, WebDAV, etc.) into a unified Document model that RAGFlow can parse and index.

Supported Sources

Supported sources are defined in the FileSource enum common/constants.py141-179 and DocumentSource enum common/data_source/config.py43-79 They include:

• Storage: S3, Google Cloud Storage, OCI Storage, R2, Dropbox, Box, Seafile, WebDAV, OneDrive, Azure Blob.
• Collaboration: Notion, Confluence, Jira, Airtable, Asana, Zendesk, Moodle, SharePoint, Salesforce.
• Communication: Slack, Discord, Gmail, IMAP, Outlook, Teams.
• Development: GitHub, GitLab, Bitbucket.
• Databases: MySQL, PostgreSQL, BigQuery, DingTalk AI Table.
• Web: RSS, REST API.

System Architecture and Data Flow

The connector system bridges the gap between external API spaces and RAGFlow's internal storage and task execution layers.

Code Entity Mapping

The following diagram maps the logical components to their specific code implementations across the React frontend, the Python backend, and the Go service layer.

Diagram: Connector System Code Entities

Sources: web/src/pages/user-setting/data-source/constant/index.tsx18-54 api/db/services/connector_service.py38-40 rag/svr/sync_data_source.py110-122 internal/service/connector.go85-96 internal/handler/connector.go52-63

Synchronization Pipeline

The synchronization process is managed by rag/svr/sync_data_source.py. It runs as a background worker that polls for scheduled tasks. The worker uses an asyncio.Semaphore named task_limiter to control concurrency, defaulting to 5 concurrent tasks rag/svr/sync_data_source.py90

Diagram: Sync Pipeline Data Flow

Sources: rag/svr/sync_data_source.py152-180 api/db/services/connector_service.py94-127 rag/svr/sync_data_source.py90

Core Components

1. The Connector Interface and Implementation

Every connector must implement or extend the logic found in common/data_source/.

• RDBMS Connector: Supports MySQL, PostgreSQL, and BigQuery. It reads rows from custom SQL queries and converts them to documents using an id_column for stable identification common/data_source/rdbms_connector.py34-46 It includes a _sanitize_query method to handle SQL pasted from markdown fences common/data_source/rdbms_connector.py100-120
• Confluence Connector: Handles Cloud and On-premise instances. It uses CQL expansions to fetch page content and handles complex OAuth2 token renewal using a Redis-backed distributed lock mechanism common/data_source/confluence_connector.py142-191
• Blob Storage: Specialized base classes in common/data_source/blob_connector.py handle S3-compatible APIs, including R2, GCS, and OCI common/data_source/config.py33-41
• OpenDAL Integration: RAGFlow utilizes opendal for abstract storage access, supporting schemes like mysql for metadata and binary storage rag/utils/opendal_conn.py23-43

2. Database Services

ConnectorService manages the lifecycle of data source connections in the MySQL database.

• Scheduling: ConnectorService.schedule_tasks() determines the poll_range_start and creates SYNC or PRUNE tasks api/db/services/connector_service.py94-127
• Cleanup: cleanup_stale_documents_for_task() removes documents that no longer exist in the source platform by comparing retain_doc_ids against existing document headers api/db/services/connector_service.py149-184
• Rebuild: ConnectorService.rebuild() allows for a complete re-indexing by deleting all documents associated with a connector and scheduling a fresh sync api/db/services/connector_service.py134-147

3. Authentication & OAuth

RAGFlow handles complex OAuth2 flows for services like Google Drive, Gmail, Box, and Confluence.

• Token Refresh: Connectors like Confluence implement _renew_credentials which checks if tokens are halfway to expiration and refreshes them via Atlassian's OAuth2 endpoints common/data_source/confluence_connector.py142-191
• Go-side OAuth Implementation: The Go backend provides specialized handlers for OAuth flows, including StartGoogleWebOAuth and StartBoxWebOAuth internal/service/connector.go118-127 internal/service/connector.go188-198 It uses Redis for temporary flow state storage with a TTL of 15 minutes internal/service/connector.go49
• Security: _redact_mailbox masks sensitive UPNs and email addresses in logs rag/svr/sync_data_source.py93-107 _redact_sensitive_url_params ensures credentials aren't leaked during HTTP logging common/http_client.py57-85

4. Frontend Integration

The frontend uses a DataSourceKey enum to manage configuration for 30+ sources web/src/pages/user-setting/data-source/constant/index.tsx18-54

• Feature Visibility: DataSourceFeatureVisibilityMap determines which sources support advanced features like syncDeletedFiles web/src/pages/user-setting/data-source/constant/index.tsx62-167
• Iconography: Centralized generateDataSourceInfo function provides UI metadata including Lucide icons and SVG icons for each source web/src/pages/user-setting/data-source/constant/index.tsx180-240

Configuration Constants

Connectors are governed by several environment variables and constants in common/data_source/config.py:

• INDEX_BATCH_SIZE: Number of documents processed in one batch (Default: 2) common/data_source/config.py121
• REQUEST_TIMEOUT_SECONDS: Global timeout for external API requests (Default: 60s) common/data_source/config.py18
• CONTINUE_ON_CONNECTOR_FAILURE: If true, a single failed document won't stop the entire sync task common/data_source/config.py153
• CONFLUENCE_TIMEZONE_OFFSET: Adjusts timestamp comparisons for Confluence sync common/data_source/config.py183

Sources: rag/svr/sync_data_source.py api/db/services/connector_service.py common/data_source/config.py common/constants.py web/src/pages/user-setting/data-source/constant/index.tsx common/data_source/rdbms_connector.py common/data_source/confluence_connector.py internal/service/connector.go internal/handler/connector.go rag/utils/opendal_conn.py common/http_client.py