Data Connectors For Ai Agents MCP Server

Data Connectors

Production-ready, reusable data connector libraries for AI applications and data processing pipelines.

Key Features:

• PostgreSQL, S3-compatible, GCP Storage, SharePoint, OneDrive connectors
• Enterprise-ready (proxy support, CA certificates)
• Type-hinted and well-documented
• Explicit configuration injection (no hidden state)
• Safe for batch jobs, CI/CD, containers, Airflow

---

🚀 Quick Start (5 Minutes)

1. Install Package

# Local development (editable)
cd /path/to/your-project
pip install -e /path/to/data-connectors

# Production (git SSH)
pip install git+ssh://git@github.com/nagarjunr/data-connectors-for-ai-agents.git@v0.1.0

# Install specific connectors only
pip install -e /path/to/data-connectors[postgres,s3]

2. Create Connector Module

File: src/connectors/__init__.py

"""Centralized connector initialization (singleton pattern)."""
from typing import Optional
import os
from dotenv import load_dotenv

# Import connectors you need
from modules.postgres import PostgresClient
from modules.s3_bucket import S3Client
from modules.gcp_bucket import GCPBucketClient
from modules.sharepoint import SharePointClient
from modules.onedrive import OneDriveClient

load_dotenv()

# Singleton instances
_pg_client: Optional[PostgresClient] = None
_s3_client: Optional[S3Client] = None
_gcp_client: Optional[GCPBucketClient] = None
_sharepoint_client: Optional[SharePointClient] = None
_onedrive_client: Optional[OneDriveClient] = None


def get_postgres_client() -> PostgresClient:
    """Get PostgreSQL client."""
    global _pg_client
    if _pg_client is None:
        _pg_client = PostgresClient(
            host=os.getenv("POSTGRES_HOST", "localhost"),
            port=int(os.getenv("POSTGRES_PORT", "5432")),
            database=os.getenv("POSTGRES_DATABASE", "your_database"),
            user=os.getenv("POSTGRES_USER"),
            password=os.getenv("POSTGRES_PASSWORD"),
        )
    return _pg_client


def get_s3_client() -> S3Client:
    """Get S3-compatible client."""
    global _s3_client
    if _s3_client is None:
        _s3_client = S3Client(
            endpoint_url=os.getenv("S3_ENDPOINT"),
            access_key=os.getenv("AWS_ACCESS_KEY_ID"),
            secret_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
            region=os.getenv("AWS_DEFAULT_REGION", "us-east-1"),
        )
    return _s3_client


def get_gcp_client() -> GCPBucketClient:
    """Get GCP Storage client."""
    global _gcp_client
    if _gcp_client is None:
        _gcp_client = GCPBucketClient(
            bucket_name=os.getenv("GCP_BUCKET_NAME"),
        )
    return _gcp_client


def get_sharepoint_client() -> SharePointClient:
    """Get SharePoint client."""
    global _sharepoint_client
    if _sharepoint_client is None:
        _sharepoint_client = SharePointClient(
            site_id=os.getenv("SITE_ID"),
            client_id=os.getenv("CLIENT_ID"),
            client_secret=os.getenv("CLIENT_SECRET"),
            tenant_id=os.getenv("TENANT_ID"),
        )
    return _sharepoint_client


def get_onedrive_client() -> OneDriveClient:
    """Get OneDrive client."""
    global _onedrive_client
    if _onedrive_client is None:
        _onedrive_client = OneDriveClient(
            client_id=os.getenv("CLIENT_ID"),
            client_secret=os.getenv("CLIENT_SECRET"),
            tenant_id=os.getenv("TENANT_ID"),
            directory=os.getenv("ONEDRIVE_DIRECTORY"),  # Optional
        )
    return _onedrive_client


def close_all_connections():
    """Close all connections on shutdown."""
    global _pg_client, _s3_client, _gcp_client, _sharepoint_client, _onedrive_client
    for client in [_pg_client, _s3_client, _gcp_client, _sharepoint_client, _onedrive_client]:
        if client:
            client.close()

3. Add Environment Variables

File: .env

# PostgreSQL
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_DATABASE=your_database
POSTGRES_USER=your_user
POSTGRES_PASSWORD=your_password

# S3-Compatible Object Storage
AWS_ACCESS_KEY_ID=your_access_key
AWS_SECRET_ACCESS_KEY=your_secret_key
S3_ENDPOINT=https://your-s3-endpoint.example.com:9021
S3_NAMESPACE=your-namespace  # Optional, only needed for some S3-compatible providers
AWS_DEFAULT_REGION=us-east-1

# GCP Storage
GCP_BUCKET_NAME=your-gcp-bucket-name

# Azure AD / Microsoft Graph (for SharePoint & OneDrive)
CLIENT_ID=your_client_id
CLIENT_SECRET=your_client_secret
TENANT_ID=your-tenant-id

# SharePoint Specific
SITE_ID=your-site-id

# OneDrive Specific
ONEDRIVE_DIRECTORY=Documents  # Optional: default directory

4. Use in Your Code

from src.connectors import (
    get_postgres_client,
    get_s3_client,
    get_gcp_client,
    get_sharepoint_client,
    get_onedrive_client,
    close_all_connections,
)

# PostgreSQL
pg = get_postgres_client()
results = pg.query("SELECT * FROM users LIMIT 10")

# S3-compatible
s3 = get_s3_client()
s3.upload_file("local.txt", "bucket-name", "remote/path/file.txt")
files = s3.list_objects("bucket-name", prefix="documents/")

# GCP Storage
gcp = get_gcp_client()
gcp.upload("/tmp/report.pdf", "reports/monthly.pdf")

# SharePoint
sp = get_sharepoint_client()
sp.download("Shared Documents/file.pdf", "/tmp/file.pdf")

# OneDrive
od = get_onedrive_client()
od.list("Documents")

# Cleanup on shutdown (FastAPI example)
# @app.on_event("shutdown")
# async def shutdown():
#     close_all_connections()

---

📦 Available Connectors

---

📚 Documentation

For Users

• Deployment Guide - Complete deployment guide from local dev to OpenShift/Kubernetes

For Contributors

• Development Guide - Development setup, code quality standards, and contributing guidelines

Quick Development Commands

# Format code
make format

# Lint and auto-fix
make lint-fix

# Type check
make typecheck

# Run all checks
make check

---

🎯 Common Use Cases

Use Case 1: Load Documents from S3

from src.connectors import get_s3_client
from pathlib import Path

def load_documents():
    s3 = get_s3_client()
    bucket = os.getenv("S3_BUCKET_NAME")
    files = s3.list_objects(bucket, prefix="documents/")

    documents = []
    for file_key in files:
        local_path = f"/tmp/{Path(file_key).name}"
        s3.download_file(bucket, file_key, local_path)
        with open(local_path, 'r') as f:
            documents.append(f.read())

    return documents

Use Case 2: Save Results to PostgreSQL

from src.connectors import get_postgres_client

def save_result(data: dict):
    pg = get_postgres_client()

    query = """
        INSERT INTO results (category, answer, status)
        VALUES (%s, %s, %s)
        RETURNING id
    """
    result_id = pg.execute(query, (data['category'], data['answer'], data['status']))
    return result_id

Use Case 3: Backup to GCP

from src.connectors import get_gcp_client
from pathlib import Path

def backup_files(source_dir: str):
    gcp = get_gcp_client()

    for file_path in Path(source_dir).rglob("*"):
        if file_path.is_file():
            gcp_path = f"backups/{file_path.name}"
            gcp.upload(str(file_path), gcp_path)

---

🏗️ Repository Structure

data-connectors/
├── README.md              # This file - Quick start and overview
├── docs/
│   ├── INTEGRATION.md     # Integration guide for using connectors
│   ├── DEPLOYMENT.md      # Complete deployment guide (local to OpenShift)
│   └── DEVELOPMENT.md     # Development setup and guidelines
├── pyproject.toml         # Python project configuration
├── requirements-dev.txt   # Development dependencies
├── Makefile               # Development commands (format, lint, typecheck)
├── examples/              # Runnable reference examples
│   ├── s3_bucket/
│   ├── sharepoint/
│   ├── onedrive/
│   ├── gcp_bucket/
│   └── postgres/
└── modules/
    ├── _common/           # Shared base classes and auth helpers
    ├── s3_bucket/           # S3-compatible connector
    ├── sharepoint/       # SharePoint connector
    ├── onedrive/         # OneDrive connector
    ├── gcp_bucket/       # GCP bucket connector
    └── postgres/         # PostgreSQL connector

---

🔐 OpenShift Quick Setup

Note: Deploy keys may be disabled depending on your Git host policy. Use Personal Access Tokens instead.

Step 1: Create GitHub Personal Access Token

1. Go to: GitHub Settings - Personal Access Tokens
2. Click "Generate new token (classic)"
3. Select repo scope (Full control of private repositories)
4. Set expiration (90 days recommended)
5. Copy token immediately - you won't see it again

Step 2: Create Secret in OpenShift

oc login --server=https://api.your-openshift-cluster.example.com:6443 --token=<your-token>
oc project your-project

oc create secret generic data-connectors-github-pat \
  --from-literal=username=<your-github-username> \
  --from-literal=password=ghp_xxxxxxxxxxxxxxxxxxxxxxxxxxxxx \
  --type=kubernetes.io/basic-auth

oc annotate secret data-connectors-github-pat \
  'build.openshift.io/source-secret-match-uri-1=https://github.com/*'

For complete OpenShift deployment instructions, see Deployment Guide.

---

🐛 Troubleshooting

Import Error

# Problem: ModuleNotFoundError: No module named 'modules'
# Solution:
pip install -e /path/to/data-connectors
pip list | grep data-connectors

Connection Error

# Problem: Connection refused to PostgreSQL/S3
# Solution:
# 1. Check .env variables
# 2. Verify services are running
# 3. Test connection:
psql -h localhost -U user -d db_name

SSL Certificate Error

# Problem: SSLError: certificate verify failed
# Solution:
export SSL_CERT_FILE=/path/to/ca-bundle.pem
export REQUESTS_CA_BUNDLE=$SSL_CERT_FILE

For more troubleshooting help, see the Deployment Guide.

---

🎓 Design Principles

• ✅ Explicit configuration: All settings injected via constructor parameters
• ✅ No hidden state: No environment variable loading inside modules
• ✅ Deterministic behavior: Same input = same output, always
• ✅ Easy testing: Pure functions, dependency injection, no side effects
• ✅ Production-ready: Designed for batch jobs, CI/CD, containers, Airflow
• ✅ Type-safe: Full type hints for better IDE support and error detection

---

🆘 Getting Help

• Issues: GitHub Issues
• Integration Guide: docs/INTEGRATION.md
• Deployment Guide: docs/DEPLOYMENT.md
• Development Guide: docs/DEVELOPMENT.md

---

📋 Checklist for New Projects

• Install data-connectors package
• Create src/connectors/__init__.py
• Add environment variables to .env
• Test connections locally
• Add to requirements.txt / requirements.prod.txt
• For OpenShift: Create PAT and secrets
• Test deployment in staging
• Deploy to production

---

License

MIT — see LICENSE.