Add 5 database connectors, fix SQL injection, and standardize agent result models

.env.example

@@ -1 +1,31 @@
-ANTHROPIC_API_KEY=your_api_key_here
+ANTHROPIC_API_KEY=your_api_key_here
+
+# Azure Databricks
+DATABRICKS_HOST=adb-xxxxx.azuredatabricks.net
+DATABRICKS_TOKEN=dapi...
+
+# Snowflake
+SNOWFLAKE_ACCOUNT=xy12345.eu-west-1
+
+# PostgreSQL
+POSTGRES_HOST=localhost
+
+# MySQL
+MYSQL_HOST=localhost
+
+# MongoDB
+MONGODB_URI=mongodb://localhost:27017
+MONGODB_DATABASE=your_database_name
+
+# Amazon Redshift
+REDSHIFT_HOST=cluster.abc123.eu-west-1.redshift.amazonaws.com
+
+# Elasticsearch
+ELASTICSEARCH_HOST=localhost
+ELASTICSEARCH_PORT=9200
+
+# DuckDB
+DUCKDB_FILE_PATH=/path/to/my.duckdb
+
+# Microsoft Fabric
+FABRIC_SERVER=xyz.datawarehouse.fabric.microsoft.com

README.md

@@ -217,7 +217,7 @@ Connect directly to your database — agents fetch a table and run the full pipe
 <br/>
 <img src="docs/images/connectors1.png" alt="Database Connectors Detail" width="900"/>
 <br/>
-<em>5 database connectors — Azure Databricks, Snowflake, PostgreSQL, MySQL, BigQuery</em>
+<em>10 database connectors — Azure Databricks, Snowflake, PostgreSQL, MySQL, BigQuery, MongoDB, Redshift, DuckDB, Microsoft Fabric, Elasticsearch</em>
 </div>
 
 ---
@@ -352,7 +352,12 @@ multi-agent-data-pipeline/
 │   │   ├── snowflake_conn.py    # Snowflake
 │   │   ├── postgres.py          # PostgreSQL
 │   │   ├── mysql.py             # MySQL
-│   │   └── bigquery.py          # BigQuery
+│   │   ├── bigquery.py          # BigQuery
+│   │   ├── mongodb.py           # MongoDB
+│   │   ├── redshift.py          # Amazon Redshift
+│   │   ├── duckdb_conn.py       # DuckDB
+│   │   ├── fabric.py            # Microsoft Fabric
+│   │   └── elasticsearch_conn.py # Elasticsearch
 │   ├── models.py                # Pydantic schemas
 │   └── pipeline.py              # Orchestrator
 ├── demo/
@@ -517,6 +522,78 @@ df = fetch_table(
 )
 ```
 
+#### MongoDB
+
+```python
+from src.connectors.mongodb import fetch_collection
+
+df = fetch_collection(
+    uri="mongodb://localhost:27017",
+    database="my_database",
+    collection="my_collection",
+    limit=1000
+)
+```
+
+#### Amazon Redshift
+
+```python
+from src.connectors.redshift import fetch_table
+
+df = fetch_table(
+    host="cluster.abc123.eu-west-1.redshift.amazonaws.com",
+    port=5439,
+    database="dev",
+    user="awsuser",
+    password="my_password",
+    table="my_table"
+)
+```
+
+#### DuckDB
+
+```python
+from src.connectors.duckdb_conn import fetch_table
+
+df = fetch_table(
+    filepath="/path/to/my.duckdb",
+    table="my_table",
+    limit=1000
+)
+```
+
+#### Microsoft Fabric
+
+> Requires [ODBC Driver 18 for SQL Server](https://learn.microsoft.com/en-us/sql/connect/odbc/download-odbc-driver-for-sql-server) installed at the OS level.
+
+```python
+from src.connectors.fabric import fetch_table
+
+df = fetch_table(
+    server="xyz.datawarehouse.fabric.microsoft.com",
+    database="my_warehouse",
+    user="user@org.com",
+    password="my_password",
+    table="my_table"
+)
+```
+
+#### Elasticsearch
+
+```python
+from src.connectors.elasticsearch_conn import fetch_index
+
+df = fetch_index(
+    host="localhost",
+    port=9200,
+    index="my_index",
+    username="elastic",   # optional
+    password="my_password",  # optional
+    use_ssl=False,
+    limit=1000
+)
+```
+
 ---
 
 ### Connector Status
@@ -528,10 +605,11 @@ df = fetch_table(
 | PostgreSQL | User/Pass | ✅ | ✅ | Stable |
 | MySQL | User/Pass | ✅ | ✅ | Stable |
 | BigQuery | Service Account JSON | ✅ | ✅ | Stable |
-| MongoDB | — | 🔜 | 🔜 | Planned |
-| Redshift | — | 🔜 | 🔜 | Planned |
-| DuckDB | — | 🔜 | 🔜 | Planned |
-| Microsoft Fabric | — | 🔜 | 🔜 | Planned |
+| MongoDB | URI | ✅ | ✅ | Stable |
+| Amazon Redshift | User/Pass | ✅ | ✅ | Stable |
+| DuckDB | File path | ✅ | ✅ | Stable |
+| Microsoft Fabric | User/Pass | ✅ | ✅ | Requires ODBC Driver 18 |
+| Elasticsearch | Optional User/Pass | ✅ | ✅ | Stable |
 
 > Want to add a connector? See [Contributing](#contributing)
 
@@ -693,11 +771,14 @@ We want to support every major database. Next targets:
 
 | Database | Difficulty | Issue |
 |----------|-----------|-------|
-| MongoDB | Medium | #1 |
-| Redshift | Easy | #2 |
-| DuckDB | Easy | #3 |
-| Microsoft Fabric | Medium | #4 |
-| Elasticsearch | Hard | #5 |
+| MongoDB | Medium | #1 ✅ |
+| Redshift | Easy | #2 ✅ |
+| DuckDB | Easy | #3 ✅ |
+| Microsoft Fabric | Medium | #4 ✅ |
+| Elasticsearch | Hard | #5 ✅ |
+| Oracle DB | Medium | #6 |
+| CockroachDB | Easy | #7 |
+| ClickHouse | Medium | #8 |
 
 #### ☁️ Cloud Implementations
 Deploy this on your cloud and contribute the implementation:
@@ -802,13 +883,17 @@ Follow this pattern — every agent has the same structure:
 ```python
 # src/agents/your_agent.py
 
+from pydantic import BaseModel, Field
+from typing import List
+
 SYSTEM_PROMPT = """You are a [role] agent.
 Respond ONLY with valid JSON. No markdown. No explanation.
 JSON format: { ... }"""
 
-class YourAgentResult:
-    def __init__(self, **kwargs): ...
-    def model_dump(self): return self.__dict__
+class YourAgentResult(BaseModel):
+    some_field: str = "default"
+    some_list: List[str] = Field(default_factory=list)
+    some_count: int = 0
 
 def run(data: str, context: int) -> YourAgentResult:
     response = client.messages.create(...)
@@ -867,7 +952,7 @@ tests/test_pipeline.py::TestCSVLoading::test_demo_csv_has_rows PASSED
 | Validation | Pydantic v2 |
 | CLI | Typer + Rich |
 | UI | Streamlit |
-| Connectors | Databricks SDK, Snowflake, psycopg2, mysql-connector, BigQuery |
+| Connectors | Databricks SDK, Snowflake, psycopg2, mysql-connector, BigQuery, pymongo, redshift-connector, duckdb, elasticsearch, pyodbc |
 | Testing | pytest |
 | Packaging | pyproject.toml |
 
@@ -877,15 +962,16 @@ tests/test_pipeline.py::TestCSVLoading::test_demo_csv_has_rows PASSED
 
 - [x] CSV pipeline — 5 agents
 - [x] PDF intelligence — 5 agents
-- [x] Database connectors — 5 databases
+- [x] Database connectors — 10 databases
 - [x] Streamlit UI — dark theme
 - [x] CLI entrypoint
 - [x] JSON export
+- [x] MongoDB connector
+- [x] Redshift connector
+- [x] DuckDB connector
+- [x] Microsoft Fabric connector
+- [x] Elasticsearch connector
 - [ ] pip package — `pip install multi-agent-data-pipeline`
-- [ ] MongoDB connector
-- [ ] Redshift connector
-- [ ] DuckDB connector
-- [ ] Microsoft Fabric connector
 - [ ] Async parallel agent execution
 - [ ] Agent memory — learn from past runs
 - [ ] Webhook support — trigger via HTTP

app.py

@@ -630,7 +630,7 @@ def run_pipeline_ui(df):
 
     db_type = st.selectbox(
         "Database",
-        ["Azure Databricks", "Snowflake", "PostgreSQL", "MySQL", "BigQuery"],
+        ["Azure Databricks", "Snowflake", "PostgreSQL", "MySQL", "BigQuery", "MongoDB", "Microsoft Fabric", "Amazon Redshift", "DuckDB", "Elasticsearch"],
         label_visibility="collapsed"
     )
 
@@ -733,6 +733,121 @@ def run_pipeline_ui(df):
             else:
                 st.warning("Please fill all fields")
 
+    elif db_type == "MongoDB":
+        col1, col2 = st.columns(2)
+        with col1:
+            uri = st.text_input("Connection URI", placeholder="mongodb://localhost:27017")
+            collection = st.text_input("Collection", placeholder="my_collection")
+        with col2:
+            database = st.text_input("Database", placeholder="my_database")
+            limit = st.number_input("Row limit", min_value=1, max_value=100000, value=1000)
+
+        if st.button("🔌 Connect & Fetch Collection"):
+            if uri and database and collection:
+                try:
+                    from src.connectors.mongodb import fetch_collection
+                    with st.spinner("Connecting to MongoDB..."):
+                        df = fetch_collection(uri, database, collection, int(limit))
+                    st.success(f"Connected — {len(df)} documents fetched from {database}.{collection}")
+                    st.dataframe(df, use_container_width=True, height=240)
+                except Exception as e:
+                    st.error(f"Connection failed: {e}")
+            else:
+                st.warning("Please fill all fields")
+
+    elif db_type == "Microsoft Fabric":
+        col1, col2 = st.columns(2)
+        with col1:
+            server = st.text_input("Server", placeholder="xyz.datawarehouse.fabric.microsoft.com")
+            database = st.text_input("Database", placeholder="my_warehouse")
+            table = st.text_input("Table", placeholder="my_table")
+        with col2:
+            user = st.text_input("Username", placeholder="user@org.com")
+            password = st.text_input("Password", type="password")
+
+        if st.button("🔌 Connect & Fetch Table"):
+            if server and database and user and password and table:
+                try:
+                    from src.connectors.fabric import fetch_table
+                    with st.spinner("Connecting to Microsoft Fabric..."):
+                        df = fetch_table(server, database, user, password, table)
+                    st.success(f"Connected — {len(df)} rows fetched from {table}")
+                    st.dataframe(df, use_container_width=True, height=240)
+                except Exception as e:
+                    st.error(f"Connection failed: {e}")
+            else:
+                st.warning("Please fill all fields")
+
+    elif db_type == "Amazon Redshift":
+        col1, col2 = st.columns(2)
+        with col1:
+            host = st.text_input("Host", placeholder="cluster.abc123.eu-west-1.redshift.amazonaws.com")
+            database = st.text_input("Database", placeholder="dev")
+            table = st.text_input("Table", placeholder="my_table")
+        with col2:
+            port = st.text_input("Port", value="5439")
+            user = st.text_input("Username", placeholder="awsuser")
+            password = st.text_input("Password", type="password")
+
+        if st.button("🔌 Connect & Fetch Table"):
+            if host and database and user and password and table:
+                try:
+                    from src.connectors.redshift import fetch_table
+                    with st.spinner("Connecting to Redshift..."):
+                        df = fetch_table(host, int(port), database, user, password, table)
+                    st.success(f"Connected — {len(df)} rows fetched from {table}")
+                    st.dataframe(df, use_container_width=True, height=240)
+                except Exception as e:
+                    st.error(f"Connection failed: {e}")
+            else:
+                st.warning("Please fill all fields")
+
+    elif db_type == "DuckDB":
+        col1, col2 = st.columns(2)
+        with col1:
+            filepath = st.text_input("Database file path", placeholder="/path/to/my.duckdb")
+            table = st.text_input("Table", placeholder="my_table")
+        with col2:
+            limit = st.number_input("Row limit", min_value=1, max_value=100000, value=1000)
+
+        if st.button("🔌 Connect & Fetch Table"):
+            if filepath and table:
+                try:
+                    from src.connectors.duckdb_conn import fetch_table
+                    with st.spinner("Connecting to DuckDB..."):
+                        df = fetch_table(filepath, table, int(limit))
+                    st.success(f"Connected — {len(df)} rows fetched from {table}")
+                    st.dataframe(df, use_container_width=True, height=240)
+                except Exception as e:
+                    st.error(f"Connection failed: {e}")
+            else:
+                st.warning("Please fill all fields")
+
+    elif db_type == "Elasticsearch":
+        col1, col2 = st.columns(2)
+        with col1:
+            es_host = st.text_input("Host", placeholder="localhost")
+            es_index = st.text_input("Index", placeholder="my_index")
+            es_limit = st.number_input("Row limit", min_value=1, max_value=10000, value=1000)
+        with col2:
+            es_port = st.text_input("Port", value="9200")
+            es_user = st.text_input("Username (optional)", placeholder="elastic")
+            es_password = st.text_input("Password (optional)", type="password")
+            es_ssl = st.checkbox("Use SSL (HTTPS)")
+
+        if st.button("🔌 Connect & Fetch Index"):
+            if es_host and es_index:
+                try:
+                    from src.connectors.elasticsearch_conn import fetch_index
+                    with st.spinner("Connecting to Elasticsearch..."):
+                        df = fetch_index(es_host, int(es_port), es_index, es_user, es_password, es_ssl, int(es_limit))
+                    st.success(f"Connected — {len(df)} documents fetched from {es_index}")
+                    st.dataframe(df, use_container_width=True, height=240)
+                except Exception as e:
+                    st.error(f"Connection failed: {e}")
+            else:
+                st.warning("Please fill Host and Index")
+
     elif db_type == "BigQuery":
         col1, col2 = st.columns(2)
         with col1:

main.py

@@ -3,7 +3,7 @@
 import os
 from rich.console import Console
 from rich.panel import Panel
-from src.pipeline import run_pipeline
+from src.pipeline import run_pipeline, run_pipeline_mongo
 
 app = typer.Typer()
 console = Console()
@@ -36,5 +36,28 @@ def run(
             json.dump(result.model_dump(), f, indent=2)
         console.print(f"[green]→ Results saved to {output}[/green]")
 
+@app.command("run-mongo")
+def run_mongo(
+    uri: str = typer.Option(..., "--uri", help="MongoDB connection URI (e.g. mongodb://localhost:27017)"),
+    database: str = typer.Option(..., "--database", "-d", help="Database name"),
+    collection: str = typer.Option(..., "--collection", "-c", help="Collection name"),
+    limit: int = typer.Option(1000, "--limit", "-l", help="Max documents to fetch"),
+    output: str = typer.Option(None, "--output", "-o", help="Save results to JSON file"),
+):
+    """
+    Run the pipeline on a MongoDB collection.
+
+    Example:
+        python main.py run-mongo --uri mongodb://localhost:27017 --database mydb --collection orders
+    """
+    result = run_pipeline_mongo(uri, database, collection, limit)
+
+    if output:
+        import json
+        with open(output, "w") as f:
+            json.dump(result.model_dump(), f, indent=2)
+        console.print(f"[green]→ Results saved to {output}[/green]")
+
+
 if __name__ == "__main__":
     app()