This guide shows you how to add a new data source connector to Docling Pipelines using the hexagonal architecture adapter pattern.
Docling Pipelines uses a hexagonal architecture with adapters for data sources. Each connector consists of:
DocumentSourcePort interface with required methods:
fetch_documents(): Async generator yielding Document objectstest_connection(): Validates connectivity to the sourceget_config_schema(): Returns the Pydantic config model classbuild_config_from_operator_params(): Maps operator params to adapter config@register_source_adapter decorator (no parameters needed)SOURCE_NAME, SOURCE_DISPLAY_NAME, SOURCE_DESCRIPTION, SOURCE_VERSIONCreate your connector in this location:
src/docpipe/core/operators/ingest/adapters/outbound/sources/
└── your_connector/
├── __init__.py # Export adapter and config
├── adapter.py # Main adapter implementation
├── config.py # Pydantic configuration model
└── README.md # Required connector documentation
+--------------------------------------+
| Flow JSON config |
| provider, connection, credentials |
+------------------+-------------------+
|
v
+--------------------------------------+
| IngestSourceOperator |
| builds adapter-specific config model |
+------------------+-------------------+
|
v
+--------------------------------------+
| DocumentSourcePort |
| fetch_documents / test_connection |
+------------------+-------------------+
|
+------------------+-------------------+
| |
v v
+-------------------------------+ +-------------------------------+
| YourConnectorSourceAdapter | | SourceAdapterFactory |
| adapter.py |<-----| @register_source_adapter |
+---------------+---------------+ +-------------------------------+
|
v
+-------------------------------+
| YourConnectorConfig |
| config.py (Pydantic model) |
+---------------+---------------+
|
v
+-------------------------------+
| Provider SDK / external API |
| aiohttp, boto3, msal, etc. |
+---------------+---------------+
|
v
+-------------------------------+
| Domain Document instances |
| yielded back to the operator |
+-------------------------------+
Before writing any code, identify what credentials your connector needs by consulting the API documentation:
Check Google Drive API docs:
Required:
credentials_path: Path to OAuth credentials JSON filescopes: OAuth scopes (e.g., drive.readonly)Optional:
token_path: Where to store OAuth tokenfolder_id: Specific folder to ingest fromrecursive: Traverse subdirectoriesfile_extensions: Filter by file typeHow to get these:
Group your configuration fields logically:
class ConnectorConfig(BaseModel):
"""Configuration for Connector."""
# 1. Connection Parameters (how to reach the service)
host: str = Field(..., description="Service hostname")
port: int = Field(443, description="Service port")
# 2. Authentication (how to authenticate)
api_key: str = Field(..., description="API key")
# OR
username: str = Field(..., description="Username")
password: str = Field(..., description="Password")
# 3. Data Location (what data to fetch)
folder_path: str = Field("/", description="Folder to ingest")
database: str = Field(..., description="Database name")
# 4. Behavior (how to fetch)
recursive: bool = Field(True, description="Recursive traversal")
batch_size: int = Field(100, description="Batch size")
# 5. Filters (what to include/exclude)
file_extensions: List[str] = Field(default_factory=list)
max_file_size_mb: Optional[int] = None
If your connector needs an external SDK or client library, add it to the project dependency manifest before implementing the adapter.
pyproject.tomlpyproject.tomluv syncExamples:
boto3 for Amazon S3-compatible APIsgoogle-api-python-client / google-auth-* for Google integrationsmsal for Microsoft Graph authenticationaiohttp for async HTTP-based connectorsDocument any new dependency in the connector README.md and in broader project docs if the dependency changes user setup.
config.py)from pydantic import BaseModel, Field, field_validator
from typing import Optional, List
class YourConnectorConfig(BaseModel):
"""Configuration for Your Connector source."""
# Connection parameters
api_endpoint: str = Field(..., description="API endpoint URL")
folder_path: str = Field("/", description="Folder path to start from")
# Credentials
access_token: str = Field(..., description="Access token for authentication")
# Optional parameters
recursive: bool = Field(True, description="Recursively list folders")
file_types: Optional[List[str]] = Field(None, description="File types to include")
@field_validator("access_token")
@classmethod
def validate_token(cls, v: str) -> str:
"""Validate access token format."""
if not v or len(v) < 10:
raise ValueError("Invalid access token")
return v
Key Points:
BaseModel for type safetyField() with descriptions for documentation@field_validator for custom validationadapter.py)from typing import AsyncGenerator
import aiohttp
import json
from docpipe.core.operators.ingest.ports.outbound.document_source import DocumentSourcePort
from docpipe.core.operators.ingest.domain.models import Document
from docpipe.core.operators.ingest.adapters.outbound.sources.factories.source_factory import register_source_adapter
from .config import YourConnectorConfig
@register_source_adapter
class YourConnectorSourceAdapter(DocumentSourcePort[YourConnectorConfig]):
"""Adapter for Your Connector data source."""
# Metadata for connector discovery
SOURCE_NAME = "your_connector"
SOURCE_DISPLAY_NAME = "Your Connector"
SOURCE_DESCRIPTION = "Ingest documents from Your Connector"
SOURCE_VERSION = "1.0.0"
async def fetch_documents(
self, *, config: YourConnectorConfig
) -> AsyncGenerator[Document, None]:
"""
Fetch documents from Your Connector.
Args:
config: Configuration for the connector
Yields:
Document: Domain document with content and metadata
"""
async with aiohttp.ClientSession() as session:
# List files from API
files = await self._list_files(session=session, config=config)
# Download and yield each file
for file_metadata in files:
try:
# Download file content
content = await self._download_file(
session=session,
config=config,
file_metadata=file_metadata,
)
# Create domain Document
doc = Document(
id=file_metadata["id"],
name=file_metadata["name"],
source_url=file_metadata["url"],
content=content, # bytes
size=len(content),
mimetype=file_metadata.get("mime_type", "application/octet-stream"),
extension=self._get_extension(filename=file_metadata["name"]),
modified_time=file_metadata.get("modified_time"),
metadata={
"path": file_metadata.get("path"),
"custom_field": file_metadata.get("custom_field"),
# Add any connector-specific metadata
}
)
yield doc
except Exception as e:
# Log error but continue processing other files
print(f"Error processing {file_metadata['name']}: {e}")
continue
async def _list_files(
self,
*,
session: aiohttp.ClientSession,
config: YourConnectorConfig
) -> list[dict]:
"""List files from the connector API."""
url = f"{config.api_endpoint}/files/list"
headers = {"Authorization": f"Bearer {config.access_token}"}
data = {
"path": config.folder_path,
"recursive": config.recursive
}
async with session.post(url, headers=headers, json=data) as response:
response.raise_for_status()
result = await response.json()
return [entry for entry in result["entries"] if entry["type"] == "file"]
async def _download_file(
self,
*,
session: aiohttp.ClientSession,
config: YourConnectorConfig,
file_metadata: dict
) -> bytes:
"""Download file content."""
url = f"{config.api_endpoint}/files/download"
headers = {
"Authorization": f"Bearer {config.access_token}",
"File-Path": file_metadata["path"]
}
async with session.get(url, headers=headers) as response:
response.raise_for_status()
return await response.read()
def _get_extension(self, *, filename: str) -> str:
"""Extract file extension."""
return filename.split(".")[-1] if "." in filename else ""
async def test_connection(self, *, config: YourConnectorConfig) -> tuple[bool, str]:
"""
Test connection to Your Connector API.
Args:
config: Configuration for the connector
Returns:
Tuple[bool, str]: (success, message)
"""
try:
async with aiohttp.ClientSession() as session:
url = f"{config.api_endpoint}/health"
headers = {"Authorization": f"Bearer {config.access_token}"}
async with session.get(url, headers=headers) as response:
if response.status == 200:
return True, "Connection successful"
return False, f"Connection failed with status {response.status}"
except Exception as e:
return False, f"Connection error: {str(e)}"
def get_config_schema(self) -> type[YourConnectorConfig]:
"""Get the Pydantic configuration model for this source."""
return YourConnectorConfig
def build_config_from_operator_params(
self,
*,
connection_params: dict,
credentials: dict,
included_extensions: list[str] | None = None,
max_files: int | None = None,
) -> YourConnectorConfig:
"""
Build adapter-specific configuration from operator parameters.
Args:
connection_params: Connection parameters from operator config
credentials: Credentials from operator config
included_extensions: File extensions to include (optional)
max_files: Maximum number of files to fetch (optional)
Returns:
YourConnectorConfig: Adapter-specific configuration object
"""
return YourConnectorConfig(
api_endpoint=connection_params.get("api_endpoint"),
folder_path=connection_params.get("folder_path", "/"),
access_token=credentials.get("access_token"),
recursive=connection_params.get("recursive", True),
file_types=included_extensions or connection_params.get("file_types"),
)
**Parameter Naming Note:**
- Users configure file filtering using `include_filter` (comma-separated string) in their flow JSON
- The operator converts this to `included_extensions` (list) and passes it to your adapter
- Your adapter config model typically uses `file_extensions` or `file_types` as the field name
- This separation keeps the user API simple while allowing flexible internal implementation
Key Points:
@register_source_adapter decorator (no parameter - uses SOURCE_NAME)fetch_documents(), test_connection(), get_config_schema(), build_config_from_operator_params()*,) for all method parameters except self/clsDocument objects from domain.modelsDocument.content as bytes__init__.py"""Your Connector source adapter."""
from .adapter import YourConnectorSourceAdapter
from .config import YourConnectorConfig
__all__ = ["YourConnectorSourceAdapter", "YourConnectorConfig"]
Each connector should include a README.md. Do not treat this as optional.
At minimum, document:
This keeps the operator discoverable and reduces repeated setup/debugging work.
Create a JSON flow file (e.g., sample_flows/use_cases/your_connector_pipeline.json):
{
"name": "Your Connector to OpenSearch Pipeline",
"description": "Ingest documents from Your Connector and store in OpenSearch",
"nodes": [
{
"id": "ingest_your_connector",
"operator_type": "ingest_source",
"operator_params": {
"provider": "your_connector",
"connection_params": {
"api_endpoint": "https://api.yourconnector.com",
"folder_path": "/Documents",
"recursive": true
},
"credentials": {
"access_token": "${YOUR_CONNECTOR_ACCESS_TOKEN}"
}
}
},
{
"id": "chunk_documents",
"operator_type": "chunker",
"operator_params": {
"chunk_size": 512,
"chunk_overlap": 50
}
},
{
"id": "generate_embeddings",
"operator_type": "embeddings_operator",
"operator_params": {
"model_name": "nomic-embed-text"
}
},
{
"id": "store_opensearch",
"operator_type": "opensearch_operator",
"operator_params": {
"index_name": "your_connector_docs",
"dimension": 768
}
}
],
"edges": [
{"source": "ingest_your_connector", "target": "chunk_documents"},
{"source": "chunk_documents", "target": "generate_embeddings"},
{"source": "generate_embeddings", "target": "store_opensearch"}
]
}
Here’s a complete example implementing a Dropbox connector:
Note: A production Dropbox connector now ships with the project. This section remains a simplified teaching example built on raw HTTP calls; the shipped adapter uses the Dropbox SDK, supports refresh-token authentication and paginates listings. See
sources/dropbox/README.mdfor the real implementation.
dropbox/config.pyfrom pydantic import BaseModel, Field, field_validator
from typing import Optional, List
class DropboxConfig(BaseModel):
"""Configuration for Dropbox connector."""
access_token: str = Field(..., description="Dropbox access token")
folder_path: str = Field("/", description="Folder path to start from")
recursive: bool = Field(True, description="Recursively list folders")
file_types: Optional[List[str]] = Field(None, description="File types to include")
@field_validator("access_token")
@classmethod
def validate_token(cls, v: str) -> str:
if not v or len(v) < 10:
raise ValueError("Invalid access token")
return v
dropbox/adapter.pyfrom typing import AsyncGenerator, List
import aiohttp
import json
from docpipe.core.operators.ingest.ports.outbound.document_source import DocumentSourcePort
from docpipe.core.operators.ingest.domain.models import Document
from docpipe.core.operators.ingest.adapters.outbound.sources.factories.source_factory import register_source_adapter
from .config import DropboxConfig
@register_source_adapter
class DropboxSourceAdapter(DocumentSourcePort[DropboxConfig]):
"""Adapter for Dropbox."""
SOURCE_NAME = "dropbox"
SOURCE_DISPLAY_NAME = "Dropbox"
SOURCE_DESCRIPTION = "Ingest documents from Dropbox"
SOURCE_VERSION = "1.0.0"
async def fetch_documents(self, *, config: DropboxConfig) -> AsyncGenerator[Document, None]:
"""Fetch documents from Dropbox."""
async with aiohttp.ClientSession() as session:
# List files
files = await self._list_files(session, config)
# Download and yield each file
for file_metadata in files:
content = await self._download_file(session, config, file_metadata)
doc = Document(
id=file_metadata["id"],
name=file_metadata["name"],
content=content,
source_url=f"https://www.dropbox.com/home{file_metadata['path_display']}",
modified_time=file_metadata.get("server_modified"),
size=len(content),
mimetype="application/octet-stream",
extension=file_metadata["name"].split(".")[-1] if "." in file_metadata["name"] else "",
metadata={}
)
yield doc
async def _list_files(
self,
*,
session: aiohttp.ClientSession,
config: DropboxConfig
) -> list[dict]:
"""List files in Dropbox folder."""
url = "https://api.dropboxapi.com/2/files/list_folder"
headers = {"Authorization": f"Bearer {config.access_token}"}
data = {"path": config.folder_path, "recursive": config.recursive}
async with session.post(url, headers=headers, json=data) as response:
response.raise_for_status()
result = await response.json()
return [entry for entry in result["entries"] if entry[".tag"] == "file"]
async def _download_file(
self,
*,
session: aiohttp.ClientSession,
config: DropboxConfig,
file_metadata: dict
) -> bytes:
"""Download file content from Dropbox."""
url = "https://content.dropboxapi.com/2/files/download"
headers = {
"Authorization": f"Bearer {config.access_token}",
"Dropbox-API-Arg": json.dumps({"path": file_metadata["path_lower"]})
}
async with session.post(url, headers=headers) as response:
response.raise_for_status()
return await response.read()
async def test_connection(self, *, config: DropboxConfig) -> tuple[bool, str]:
"""Test connection to Dropbox."""
try:
async with aiohttp.ClientSession() as session:
url = "https://api.dropboxapi.com/2/users/get_current_account"
headers = {"Authorization": f"Bearer {config.access_token}"}
async with session.post(url, headers=headers) as response:
if response.status == 200:
return True, "Connection successful"
return False, f"Connection failed with status {response.status}"
except Exception as e:
return False, f"Connection error: {str(e)}"
def get_config_schema(self) -> type[DropboxConfig]:
"""Get the Pydantic configuration model."""
return DropboxConfig
def build_config_from_operator_params(
self,
*,
connection_params: dict,
credentials: dict,
included_extensions: list[str] | None = None,
max_files: int | None = None,
) -> DropboxConfig:
"""Build configuration from operator parameters."""
return DropboxConfig(
access_token=credentials.get("access_token"),
folder_path=connection_params.get("folder_path", "/"),
recursive=connection_params.get("recursive", True),
file_types=included_extensions or connection_params.get("file_types"),
)
tests/unit/operators/ingest/test_your_connector.py)import pytest
from unittest.mock import AsyncMock, patch, Mock
from docpipe.core.operators.ingest.adapters.outbound.sources.your_connector.adapter import YourConnectorSourceAdapter
from docpipe.core.operators.ingest.adapters.outbound.sources.your_connector.config import YourConnectorConfig
@pytest.mark.asyncio
async def test_your_connector_fetch_documents():
"""Test fetching documents from Your Connector."""
config = YourConnectorConfig(
api_endpoint="https://api.test.com",
access_token="test_token_1234567890",
folder_path="/test"
)
adapter = YourConnectorSourceAdapter()
with patch.object(adapter, '_list_files', new_callable=AsyncMock) as mock_list:
with patch.object(adapter, '_download_file', new_callable=AsyncMock) as mock_download:
# Mock API responses
mock_list.return_value = [
{
"id": "file1",
"name": "test.txt",
"path": "/test/test.txt",
"url": "https://connector.com/file1"
}
]
mock_download.return_value = b"test content"
# Fetch documents
documents = []
async for doc in adapter.fetch_documents(config=config):
documents.append(doc)
# Assertions
assert len(documents) == 1
assert documents[0].name == "test.txt"
assert documents[0].content == b"test content"
assert documents[0].size == 12
def test_config_validation():
"""Test configuration validation."""
# Valid config
config = YourConnectorConfig(
api_endpoint="https://api.test.com",
access_token="valid_token_1234567890",
folder_path="/test"
)
assert config.access_token == "valid_token_1234567890"
# Invalid token (too short)
with pytest.raises(ValueError, match="Invalid access token"):
YourConnectorConfig(
api_endpoint="https://api.test.com",
access_token="short",
folder_path="/test"
)
@pytest.mark.asyncio
async def test_your_connector_integration():
"""Integration test with IngestSourceOperator."""
from docpipe.core.operators.ingest.ingest_source import IngestSourceOperator
config = {
"provider": "your_connector",
"connection_params": {
"api_endpoint": "https://api.test.com",
"folder_path": "/test",
"recursive": True
},
"credentials": {
"access_token": "test_token_1234567890"
},
"job_id": "test-job-123",
"job_run_id": "test-run-456"
}
operator = IngestSourceOperator(config)
# Mock the adapter
with patch(
"docpipe.core.operators.ingest.adapters.outbound.sources.factories.source_factory.SourceAdapterFactory.create"
) as mock_create:
mock_adapter = Mock()
async def mock_fetch(config):
yield Mock(
id="1",
name="test.txt",
content=b"content",
source_url="https://test.com/file1",
size=7,
mimetype="text/plain",
extension="txt",
modified_time=None,
metadata={}
)
mock_adapter.fetch_documents = mock_fetch
mock_create.return_value = mock_adapter
# Execute
result_tables, metadata = operator.transform(empty_input_table)
assert len(result_tables) == 1
assert result_tables[0].num_rows > 0
Study these existing connectors for best practices:
src/docpipe/core/operators/ingest/adapters/outbound/sources/sharepoint/src/docpipe/core/operators/ingest/adapters/outbound/sources/onedrive/src/docpipe/core/operators/ingest/adapters/outbound/sources/google_drive/src/docpipe/core/operators/ingest/adapters/outbound/sources/filesystem/src/docpipe/core/operators/ingest/adapters/outbound/sources/s3/source src/docpipe_app/backend/.venv/bin/activate
export PYTHONPATH="$(pwd)/src:${PYTHONPATH}"
docling-pipelines --flow-file tests/flow_your_connector.json
source src/docpipe_app/backend/.venv/bin/activate
export PYTHONPATH="$(pwd)/src:${PYTHONPATH}"
uv run pytest tests/unit/operators/ingest/test_your_connector.py -v
bytes in Document.contentaiohttp for HTTP requests, asyncio for concurrency@register_source_adapter decorator is present (no parameters)SOURCE_NAME class attribute is defined__init__.py exports the adapter classSOURCE_NAME matches the provider value in flow JSONPYTHONPATH correctly before runninguv sync)bytes, not str_binary_content attribute for LangChain compatibilityFor questions or issues, refer to existing adapters and the Ingest Source Operator documentation.