Merge branch 'main' of https://github.com/siiddhantt/DocsGPT into pr/1930

application/api/answer/services/stream_processor.py

@@ -85,6 +85,7 @@ class StreamProcessor:
         self._configure_agent()
         self._configure_source()
         self._configure_retriever()
+        self._configure_agent()
         self._load_conversation_history()
         self._process_attachments()
 
@@ -260,6 +261,12 @@ class StreamProcessor:
             )
             self.initial_user_id = data_key.get("user")
             self.decoded_token = {"sub": data_key.get("user")}
+            if data_key.get("source"):
+                self.source = {"active_docs": data_key["source"]}
+            if data_key.get("retriever"):
+                self.retriever_config["retriever_name"] = data_key["retriever"]
+            if data_key.get("chunks") is not None:
+                self.retriever_config["chunks"] = data_key["chunks"]
         elif self.agent_key:
             data_key = self._get_data_from_api_key(self.agent_key)
             self.agent_config.update(
@@ -275,6 +282,12 @@ class StreamProcessor:
                 if self.is_shared_usage
                 else {"sub": data_key.get("user")}
             )
+            if data_key.get("source"):
+                self.source = {"active_docs": data_key["source"]}
+            if data_key.get("retriever"):
+                self.retriever_config["retriever_name"] = data_key["retriever"]
+            if data_key.get("chunks") is not None:
+                self.retriever_config["chunks"] = data_key["chunks"]
         else:
             self.agent_config.update(
                 {

application/api/connector/routes.py

@@ -0,0 +1,627 @@
+import datetime
+import json
+import logging
+
+
+from bson.objectid import ObjectId
+from flask import (
+    Blueprint,
+    current_app,
+    jsonify,
+    make_response,
+    request
+)
+from flask_restx import fields, Namespace, Resource
+
+
+
+
+from application.api.user.tasks import (
+    ingest_connector_task,
+)
+from application.core.mongo_db import MongoDB
+from application.core.settings import settings
+from application.api import api
+
+
+from application.utils import (
+    check_required_fields
+)
+
+
+from application.parser.connectors.connector_creator import ConnectorCreator
+
+
+
+mongo = MongoDB.get_client()
+db = mongo[settings.MONGO_DB_NAME]
+sources_collection = db["sources"]
+sessions_collection = db["connector_sessions"]
+
+connector = Blueprint("connector", __name__)
+connectors_ns = Namespace("connectors", description="Connector operations", path="/")
+api.add_namespace(connectors_ns)
+
+
+
+@connectors_ns.route("/api/connectors/upload")
+class UploadConnector(Resource):
+    @api.expect(
+        api.model(
+            "ConnectorUploadModel",
+            {
+                "user": fields.String(required=True, description="User ID"),
+                "source": fields.String(
+                    required=True, description="Source type (google_drive, github, etc.)"
+                ),
+                "name": fields.String(required=True, description="Job name"),
+                "data": fields.String(required=True, description="Configuration data"),
+                "repo_url": fields.String(description="GitHub repository URL"),
+            },
+        )
+    )
+    @api.doc(
+        description="Uploads connector source for vectorization",
+    )
+    def post(self):
+        decoded_token = request.decoded_token
+        if not decoded_token:
+            return make_response(jsonify({"success": False}), 401)
+        data = request.form
+        required_fields = ["user", "source", "name", "data"]
+        missing_fields = check_required_fields(data, required_fields)
+        if missing_fields:
+            return missing_fields
+        try:
+            config = json.loads(data["data"])
+            source_data = None
+            sync_frequency = config.get("sync_frequency", "never")
+
+            if data["source"] == "github":
+                source_data = config.get("repo_url")
+            elif data["source"] in ["crawler", "url"]:
+                source_data = config.get("url")
+            elif data["source"] == "reddit":
+                source_data = config
+            elif data["source"] in ConnectorCreator.get_supported_connectors():
+                session_token = config.get("session_token")
+                if not session_token:
+                    return make_response(jsonify({
+                        "success": False,
+                        "error": f"Missing session_token in {data['source']} configuration"
+                    }), 400)
+
+                file_ids = config.get("file_ids", [])
+                if isinstance(file_ids, str):
+                    file_ids = [id.strip() for id in file_ids.split(',') if id.strip()]
+                elif not isinstance(file_ids, list):
+                    file_ids = []
+
+                folder_ids = config.get("folder_ids", [])
+                if isinstance(folder_ids, str):
+                    folder_ids = [id.strip() for id in folder_ids.split(',') if id.strip()]
+                elif not isinstance(folder_ids, list):
+                    folder_ids = []
+
+                config["file_ids"] = file_ids
+                config["folder_ids"] = folder_ids
+
+                task = ingest_connector_task.delay(
+                    job_name=data["name"],
+                    user=decoded_token.get("sub"),
+                    source_type=data["source"],
+                    session_token=session_token,
+                    file_ids=file_ids,
+                    folder_ids=folder_ids,
+                    recursive=config.get("recursive", False),
+                    retriever=config.get("retriever", "classic"),
+                    sync_frequency=sync_frequency
+                )
+                return make_response(jsonify({"success": True, "task_id": task.id}), 200)
+            task = ingest_connector_task.delay(
+                source_data=source_data,
+                job_name=data["name"],
+                user=decoded_token.get("sub"),
+                loader=data["source"],
+                sync_frequency=sync_frequency
+            )
+        except Exception as err:
+            current_app.logger.error(
+                f"Error uploading connector source: {err}", exc_info=True
+            )
+            return make_response(jsonify({"success": False}), 400)
+        return make_response(jsonify({"success": True, "task_id": task.id}), 200)
+
+
+@connectors_ns.route("/api/connectors/task_status")
+class ConnectorTaskStatus(Resource):
+    task_status_model = api.model(
+        "ConnectorTaskStatusModel",
+        {"task_id": fields.String(required=True, description="Task ID")},
+    )
+
+    @api.expect(task_status_model)
+    @api.doc(description="Get connector task status")
+    def get(self):
+        task_id = request.args.get("task_id")
+        if not task_id:
+            return make_response(
+                jsonify({"success": False, "message": "Task ID is required"}), 400
+            )
+        try:
+            from application.celery_init import celery
+
+            task = celery.AsyncResult(task_id)
+            task_meta = task.info
+            print(f"Task status: {task.status}")
+            if not isinstance(
+                task_meta, (dict, list, str, int, float, bool, type(None))
+            ):
+                task_meta = str(task_meta)
+        except Exception as err:
+            current_app.logger.error(f"Error getting task status: {err}", exc_info=True)
+            return make_response(jsonify({"success": False}), 400)
+        return make_response(jsonify({"status": task.status, "result": task_meta}), 200)
+
+
+@connectors_ns.route("/api/connectors/sources")
+class ConnectorSources(Resource):
+    @api.doc(description="Get connector sources")
+    def get(self):
+        decoded_token = request.decoded_token
+        if not decoded_token:
+            return make_response(jsonify({"success": False}), 401)
+        user = decoded_token.get("sub")
+        try:
+            sources = sources_collection.find({"user": user, "type": "connector"}).sort("date", -1)
+            connector_sources = []
+            for source in sources:
+                connector_sources.append({
+                    "id": str(source["_id"]),
+                    "name": source.get("name"),
+                    "date": source.get("date"),
+                    "type": source.get("type"),
+                    "source": source.get("source"),
+                    "tokens": source.get("tokens", ""),
+                    "retriever": source.get("retriever", "classic"),
+                    "syncFrequency": source.get("sync_frequency", ""),
+                })
+        except Exception as err:
+            current_app.logger.error(f"Error retrieving connector sources: {err}", exc_info=True)
+            return make_response(jsonify({"success": False}), 400)
+        return make_response(jsonify(connector_sources), 200)
+
+
+@connectors_ns.route("/api/connectors/delete")
+class DeleteConnectorSource(Resource):
+    @api.doc(
+        description="Delete a connector source",
+        params={"source_id": "The source ID to delete"},
+    )
+    def delete(self):
+        decoded_token = request.decoded_token
+        if not decoded_token:
+            return make_response(jsonify({"success": False}), 401)
+        source_id = request.args.get("source_id")
+        if not source_id:
+            return make_response(
+                jsonify({"success": False, "message": "source_id is required"}), 400
+            )
+        try:
+            result = sources_collection.delete_one(
+                {"_id": ObjectId(source_id), "user": decoded_token.get("sub")}
+            )
+            if result.deleted_count == 0:
+                return make_response(
+                    jsonify({"success": False, "message": "Source not found"}), 404
+                )
+        except Exception as err:
+            current_app.logger.error(
+                f"Error deleting connector source: {err}", exc_info=True
+            )
+            return make_response(jsonify({"success": False}), 400)
+        return make_response(jsonify({"success": True}), 200)
+
+
+@connectors_ns.route("/api/connectors/auth")
+class ConnectorAuth(Resource):
+    @api.doc(description="Get connector OAuth authorization URL", params={"provider": "Connector provider (e.g., google_drive)"})
+    def get(self):
+        try:
+            provider = request.args.get('provider') or request.args.get('source')
+            if not provider:
+                return make_response(jsonify({"success": False, "error": "Missing provider"}), 400)
+
+            if not ConnectorCreator.is_supported(provider):
+                return make_response(jsonify({"success": False, "error": f"Unsupported provider: {provider}"}), 400)
+
+            import uuid
+            state = str(uuid.uuid4())
+            auth = ConnectorCreator.create_auth(provider)
+            authorization_url = auth.get_authorization_url(state=state)
+            return make_response(jsonify({
+                "success": True,
+                "authorization_url": authorization_url,
+                "state": state
+            }), 200)
+        except Exception as e:
+            current_app.logger.error(f"Error generating connector auth URL: {e}")
+            return make_response(jsonify({"success": False, "error": str(e)}), 500)
+
+
+@connectors_ns.route("/api/connectors/callback")
+class ConnectorsCallback(Resource):
+    @api.doc(description="Handle OAuth callback for external connectors")
+    def get(self):
+        """Handle OAuth callback for external connectors"""
+        try:
+            from application.parser.connectors.connector_creator import ConnectorCreator
+            from flask import request, redirect
+            import uuid
+
+            provider = request.args.get('provider', 'google_drive')
+            authorization_code = request.args.get('code')
+            _ = request.args.get('state')
+            error = request.args.get('error')
+
+            if error:
+                return redirect(f"/api/connectors/callback-status?status=error&message=OAuth+error:+{error}.+Please+try+again+and+make+sure+to+grant+all+requested+permissions,+including+offline+access.&provider={provider}")
+
+            if not authorization_code:
+                return redirect(f"/api/connectors/callback-status?status=error&message=Authorization+code+not+provided.+Please+complete+the+authorization+process+and+make+sure+to+grant+offline+access.&provider={provider}")
+
+            try:
+                auth = ConnectorCreator.create_auth(provider)
+                token_info = auth.exchange_code_for_tokens(authorization_code)
+
+                session_token = str(uuid.uuid4())
+                
+
+                try:
+                    credentials = auth.create_credentials_from_token_info(token_info)
+                    service = auth.build_drive_service(credentials)
+                    user_info = service.about().get(fields="user").execute()
+                    user_email = user_info.get('user', {}).get('emailAddress', 'Connected User')
+                except Exception as e:
+                    current_app.logger.warning(f"Could not get user info: {e}")
+                    user_email = 'Connected User'
+
+                sanitized_token_info = {
+                    "access_token": token_info.get("access_token"),
+                    "refresh_token": token_info.get("refresh_token"),
+                    "token_uri": token_info.get("token_uri"),
+                    "expiry": token_info.get("expiry"),
+                    "scopes": token_info.get("scopes")
+                }
+
+                user_id = request.decoded_token.get("sub") if getattr(request, "decoded_token", None) else None
+                sessions_collection.insert_one({
+                    "session_token": session_token,
+                    "user": user_id,
+                    "token_info": sanitized_token_info,
+                    "created_at": datetime.datetime.now(datetime.timezone.utc),
+                    "user_email": user_email,
+                    "provider": provider
+                })
+
+                # Redirect to success page with session token and user email
+                return redirect(f"/api/connectors/callback-status?status=success&message=Authentication+successful&provider={provider}&session_token={session_token}&user_email={user_email}")
+
+            except Exception as e:
+                current_app.logger.error(f"Error exchanging code for tokens: {str(e)}", exc_info=True)
+                return redirect(f"/api/connectors/callback-status?status=error&message=Failed+to+exchange+authorization+code+for+tokens:+{str(e)}&provider={provider}")
+
+        except Exception as e:
+            current_app.logger.error(f"Error handling connector callback: {e}")
+            return redirect(f"/api/connectors/callback-status?status=error&message=Failed+to+complete+connector+authentication:+{str(e)}.+Please+try+again+and+make+sure+to+grant+all+requested+permissions,+including+offline+access.")
+
+
+@connectors_ns.route("/api/connectors/refresh")
+class ConnectorRefresh(Resource):
+    @api.expect(api.model("ConnectorRefreshModel", {"provider": fields.String(required=True), "refresh_token": fields.String(required=True)}))
+    @api.doc(description="Refresh connector access token")
+    def post(self):
+        try:
+            data = request.get_json()
+            provider = data.get('provider')
+            refresh_token = data.get('refresh_token')
+
+            if not provider or not refresh_token:
+                return make_response(jsonify({"success": False, "error": "provider and refresh_token are required"}), 400)
+
+            auth = ConnectorCreator.create_auth(provider)
+            token_info = auth.refresh_access_token(refresh_token)
+            return make_response(jsonify({"success": True, "token_info": token_info}), 200)
+        except Exception as e:
+            current_app.logger.error(f"Error refreshing token for connector: {e}")
+            return make_response(jsonify({"success": False, "error": str(e)}), 500)
+
+
+@connectors_ns.route("/api/connectors/files")
+class ConnectorFiles(Resource):
+    @api.expect(api.model("ConnectorFilesModel", {"provider": fields.String(required=True), "session_token": fields.String(required=True), "folder_id": fields.String(required=False), "limit": fields.Integer(required=False), "page_token": fields.String(required=False)}))
+    @api.doc(description="List files from a connector provider (supports pagination)")
+    def post(self):
+        try:
+            data = request.get_json()
+            provider = data.get('provider')
+            session_token = data.get('session_token')
+            folder_id = data.get('folder_id')
+            limit = data.get('limit', 10)
+            page_token = data.get('page_token')
+            if not provider or not session_token:
+                return make_response(jsonify({"success": False, "error": "provider and session_token are required"}), 400)
+
+
+            decoded_token = request.decoded_token
+            if not decoded_token:
+                return make_response(jsonify({"success": False, "error": "Unauthorized"}), 401)
+            user = decoded_token.get('sub')
+            session = sessions_collection.find_one({"session_token": session_token, "user": user})
+            if not session:
+                return make_response(jsonify({"success": False, "error": "Invalid or unauthorized session"}), 401)
+
+            loader = ConnectorCreator.create_connector(provider, session_token)
+            documents = loader.load_data({
+                'limit': limit,
+                'list_only': True,
+                'session_token': session_token,
+                'folder_id': folder_id,
+                'page_token': page_token
+            })
+
+            files = []
+            for doc in documents[:limit]:
+                metadata = doc.extra_info
+                modified_time = metadata.get('modified_time')
+                if modified_time:
+                    date_part = modified_time.split('T')[0]
+                    time_part = modified_time.split('T')[1].split('.')[0].split('Z')[0]
+                    formatted_time = f"{date_part} {time_part}"
+                else:
+                    formatted_time = None
+
+                files.append({
+                    'id': doc.doc_id,
+                    'name': metadata.get('file_name', 'Unknown File'),
+                    'type': metadata.get('mime_type', 'unknown'),
+                    'size': metadata.get('size', None),
+                    'modifiedTime': formatted_time
+                })
+
+            next_token = getattr(loader, 'next_page_token', None)
+            has_more = bool(next_token)
+
+            return make_response(jsonify({"success": True, "files": files, "total": len(files), "next_page_token": next_token, "has_more": has_more}), 200)
+        except Exception as e:
+            current_app.logger.error(f"Error loading connector files: {e}")
+            return make_response(jsonify({"success": False, "error": f"Failed to load files: {str(e)}"}), 500)
+
+
+@connectors_ns.route("/api/connectors/validate-session")
+class ConnectorValidateSession(Resource):
+    @api.expect(api.model("ConnectorValidateSessionModel", {"provider": fields.String(required=True), "session_token": fields.String(required=True)}))
+    @api.doc(description="Validate connector session token and return user info")
+    def post(self):
+        try:
+            data = request.get_json()
+            provider = data.get('provider')
+            session_token = data.get('session_token')
+            if not provider or not session_token:
+                return make_response(jsonify({"success": False, "error": "provider and session_token are required"}), 400)
+
+
+            decoded_token = request.decoded_token
+            if not decoded_token:
+                return make_response(jsonify({"success": False, "error": "Unauthorized"}), 401)
+            user = decoded_token.get('sub')
+
+            session = sessions_collection.find_one({"session_token": session_token, "user": user})
+            if not session or "token_info" not in session:
+                return make_response(jsonify({"success": False, "error": "Invalid or expired session"}), 401)
+
+            token_info = session["token_info"]
+            auth = ConnectorCreator.create_auth(provider)
+            is_expired = auth.is_token_expired(token_info)
+
+            return make_response(jsonify({
+                "success": True,
+                "expired": is_expired,
+                "user_email": session.get('user_email', 'Connected User')
+            }), 200)
+        except Exception as e:
+            current_app.logger.error(f"Error validating connector session: {e}")
+            return make_response(jsonify({"success": False, "error": str(e)}), 500)
+
+
+@connectors_ns.route("/api/connectors/disconnect")
+class ConnectorDisconnect(Resource):
+    @api.expect(api.model("ConnectorDisconnectModel", {"provider": fields.String(required=True), "session_token": fields.String(required=False)}))
+    @api.doc(description="Disconnect a connector session")
+    def post(self):
+        try:
+            data = request.get_json()
+            provider = data.get('provider')
+            session_token = data.get('session_token')
+            if not provider:
+                return make_response(jsonify({"success": False, "error": "provider is required"}), 400)
+
+
+            if session_token:
+                sessions_collection.delete_one({"session_token": session_token})
+            
+            return make_response(jsonify({"success": True}), 200)
+        except Exception as e:
+            current_app.logger.error(f"Error disconnecting connector session: {e}")
+            return make_response(jsonify({"success": False, "error": str(e)}), 500)
+
+
+@connectors_ns.route("/api/connectors/sync")
+class ConnectorSync(Resource):
+    @api.expect(
+        api.model(
+            "ConnectorSyncModel",
+            {
+                "source_id": fields.String(required=True, description="Source ID to sync"),
+                "session_token": fields.String(required=True, description="Authentication token")
+            },
+        )
+    )
+    @api.doc(description="Sync connector source to check for modifications")
+    def post(self):
+        decoded_token = request.decoded_token
+        if not decoded_token:
+            return make_response(jsonify({"success": False}), 401)
+
+        try:
+            data = request.get_json()
+            source_id = data.get('source_id')
+            session_token = data.get('session_token')
+
+            if not all([source_id, session_token]):
+                return make_response(
+                    jsonify({
+                        "success": False,
+                        "error": "source_id and session_token are required"
+                    }), 
+                    400
+                )
+            source = sources_collection.find_one({"_id": ObjectId(source_id)})
+            if not source:
+                return make_response(
+                    jsonify({
+                        "success": False,
+                        "error": "Source not found"
+                    }), 
+                    404
+                )
+
+            if source.get('user') != decoded_token.get('sub'):
+                return make_response(
+                    jsonify({
+                        "success": False,
+                        "error": "Unauthorized access to source"
+                    }), 
+                    403
+                )
+
+            remote_data = {}
+            try:
+                if source.get('remote_data'):
+                    remote_data = json.loads(source.get('remote_data'))
+            except json.JSONDecodeError:
+                current_app.logger.error(f"Invalid remote_data format for source {source_id}")
+                remote_data = {}
+
+            source_type = remote_data.get('provider')
+            if not source_type:
+                return make_response(
+                    jsonify({
+                        "success": False,
+                        "error": "Source provider not found in remote_data"
+                    }), 
+                    400
+                )
+
+            # Extract configuration from remote_data
+            file_ids = remote_data.get('file_ids', [])
+            folder_ids = remote_data.get('folder_ids', [])
+            recursive = remote_data.get('recursive', True)
+
+            # Start the sync task
+            task = ingest_connector_task.delay(
+                job_name=source.get('name'),
+                user=decoded_token.get('sub'),
+                source_type=source_type,
+                session_token=session_token,
+                file_ids=file_ids,
+                folder_ids=folder_ids,
+                recursive=recursive,
+                retriever=source.get('retriever', 'classic'),
+                operation_mode="sync",
+                doc_id=source_id,
+                sync_frequency=source.get('sync_frequency', 'never')
+            )
+
+            return make_response(
+                jsonify({
+                    "success": True,
+                    "task_id": task.id
+                }), 
+                200
+            )
+
+        except Exception as err:
+            current_app.logger.error(
+                f"Error syncing connector source: {err}",
+                exc_info=True
+            )
+            return make_response(
+                jsonify({
+                    "success": False,
+                    "error": str(err)
+                }), 
+                400
+            )
+
+
+@connectors_ns.route("/api/connectors/callback-status")
+class ConnectorCallbackStatus(Resource):
+    @api.doc(description="Return HTML page with connector authentication status")
+    def get(self):
+        """Return HTML page with connector authentication status"""
+        try:
+            status = request.args.get('status', 'error')
+            message = request.args.get('message', '')
+            provider = request.args.get('provider', 'connector')
+            session_token = request.args.get('session_token', '')
+            user_email = request.args.get('user_email', '')
+            
+            html_content = f"""
+            <!DOCTYPE html>
+            <html>
+            <head>
+                <title>{provider.replace('_', ' ').title()} Authentication</title>
+                <style>
+                    body {{ font-family: Arial, sans-serif; text-align: center; padding: 40px; }}
+                    .container {{ max-width: 600px; margin: 0 auto; }}
+                    .success {{ color: #4CAF50; }}
+                    .error {{ color: #F44336; }}
+                </style>
+                <script>
+                    window.onload = function() {{
+                        const status = "{status}";
+                        const sessionToken = "{session_token}";
+                        const userEmail = "{user_email}";
+                        
+                        if (status === "success" && window.opener) {{
+                            window.opener.postMessage({{
+                                type: '{provider}_auth_success',
+                                session_token: sessionToken,
+                                user_email: userEmail
+                            }}, '*');
+                            
+                            setTimeout(() => window.close(), 3000);
+                        }}
+                    }};
+                </script>
+            </head>
+            <body>
+                <div class="container">
+                    <h2>{provider.replace('_', ' ').title()} Authentication</h2>
+                    <div class="{status}">
+                        <p>{message}</p>
+                        {f'<p>Connected as: {user_email}</p>' if status == 'success' else ''}
+                    </div>
+                    <p><small>You can close this window. {f"Your {provider.replace('_', ' ').title()} is now connected and ready to use." if status == 'success' else ''}</small></p>
+                </div>
+            </body>
+            </html>
+            """
+            
+            return make_response(html_content, 200, {'Content-Type': 'text/html'})
+        except Exception as e:
+            current_app.logger.error(f"Error rendering callback status page: {e}")
+            return make_response("Authentication error occurred", 500, {'Content-Type': 'text/html'})
+
+

application/api/user/routes.py

@@ -3,11 +3,12 @@ import json
 import math
 import os
 import secrets
+import tempfile
 import uuid
+import zipfile
 from functools import wraps
 from typing import Optional, Tuple
-import tempfile
-import zipfile
+
 from bson.binary import Binary, UuidRepresentation
 from bson.dbref import DBRef
 from bson.objectid import ObjectId
@@ -25,26 +26,28 @@ from pymongo import ReturnDocument
 from werkzeug.utils import secure_filename
 
 from application.agents.tools.tool_manager import ToolManager
+from application.api import api
 
 from application.api.user.tasks import (
     ingest,
+    ingest_connector_task,
     ingest_remote,
     process_agent_webhook,
     store_attachment,
 )
 from application.core.mongo_db import MongoDB
 from application.core.settings import settings
-from application.api import api
+from application.parser.connectors.connector_creator import ConnectorCreator
 from application.storage.storage_creator import StorageCreator
 from application.tts.google_tts import GoogleTTS
 from application.utils import (
     check_required_fields,
     generate_image_url,
+    num_tokens_from_string,
     safe_filename,
     validate_function_name,
     validate_required_fields,
 )
-from application.utils import num_tokens_from_string
 from application.vectorstore.vector_creator import VectorCreator
 
 storage = StorageCreator.get_storage()
@@ -72,7 +75,6 @@ try:
     users_collection.create_index("user_id", unique=True)
 except Exception as e:
     print("Error creating indexes:", e)
-
 user = Blueprint("user", __name__)
 user_ns = Namespace("user", description="User related operations", path="/")
 api.add_namespace(user_ns)
@@ -125,11 +127,9 @@ def ensure_user_doc(user_id):
         updates["agent_preferences.pinned"] = []
     if "shared_with_me" not in prefs:
         updates["agent_preferences.shared_with_me"] = []
-
     if updates:
         users_collection.update_one({"user_id": user_id}, {"$set": updates})
         user_doc = users_collection.find_one({"user_id": user_id})
-
     return user_doc
 
 
@@ -181,7 +181,6 @@ def handle_image_upload(
                     jsonify({"success": False, "message": "Image upload failed"}),
                     400,
                 )
-
     return image_url, None
 
 
@@ -295,8 +294,8 @@ class GetSingleConversation(Resource):
             )
             if not conversation:
                 return make_response(jsonify({"status": "not found"}), 404)
-
             # Process queries to include attachment names
+
             queries = conversation["queries"]
             for query in queries:
                 if "attachments" in query and query["attachments"]:
@@ -492,11 +491,11 @@ class DeleteOldIndexes(Resource):
         )
         if not doc:
             return make_response(jsonify({"status": "not found"}), 404)
-
         storage = StorageCreator.get_storage()
 
         try:
             # Delete vector index
+
             if settings.VECTOR_STORE == "faiss":
                 index_path = f"indexes/{str(doc['_id'])}"
                 if storage.file_exists(f"{index_path}/index.faiss"):
@@ -508,7 +507,6 @@ class DeleteOldIndexes(Resource):
                     settings.VECTOR_STORE, source_id=str(doc["_id"])
                 )
                 vectorstore.delete_index()
-
             if "file_path" in doc and doc["file_path"]:
                 file_path = doc["file_path"]
                 if storage.is_directory(file_path):
@@ -517,7 +515,6 @@ class DeleteOldIndexes(Resource):
                         storage.delete_file(f)
                 else:
                     storage.delete_file(file_path)
-
         except FileNotFoundError:
             pass
         except Exception as err:
@@ -525,7 +522,6 @@ class DeleteOldIndexes(Resource):
                 f"Error deleting files and indexes: {err}", exc_info=True
             )
             return make_response(jsonify({"success": False}), 400)
-
         sources_collection.delete_one({"_id": ObjectId(source_id)})
         return make_response(jsonify({"success": True}), 200)
 
@@ -567,6 +563,7 @@ class UploadFile(Resource):
         job_name = request.form["name"]
 
         # Create safe versions for filesystem operations
+
         safe_user = safe_filename(user)
         dir_name = safe_filename(job_name)
         base_path = f"{settings.UPLOAD_FOLDER}/{safe_user}/{dir_name}"
@@ -588,6 +585,7 @@ class UploadFile(Resource):
                                 zip_ref.extractall(path=temp_dir)
 
                                 # Walk through extracted files and upload them
+
                                 for root, _, files in os.walk(temp_dir):
                                     for extracted_file in files:
                                         if (
@@ -595,7 +593,6 @@ class UploadFile(Resource):
                                             == temp_file_path
                                         ):
                                             continue
-
                                         rel_path = os.path.relpath(
                                             os.path.join(root, extracted_file), temp_dir
                                         )
@@ -610,15 +607,16 @@ class UploadFile(Resource):
                                 f"Error extracting zip: {e}", exc_info=True
                             )
                             # If zip extraction fails, save the original zip file
+
                             file_path = f"{base_path}/{safe_file}"
                             with open(temp_file_path, "rb") as f:
                                 storage.save_file(f, file_path)
                     else:
                         # For non-zip files, save directly
+
                         file_path = f"{base_path}/{safe_file}"
                         with open(temp_file_path, "rb") as f:
                             storage.save_file(f, file_path)
-
             task = ingest.delay(
                 settings.UPLOAD_FOLDER,
                 [
@@ -690,7 +688,6 @@ class ManageSourceFiles(Resource):
             return make_response(
                 jsonify({"success": False, "message": "Unauthorized"}), 401
             )
-
         user = decoded_token.get("sub")
         source_id = request.form.get("source_id")
         operation = request.form.get("operation")
@@ -705,7 +702,6 @@ class ManageSourceFiles(Resource):
                 ),
                 400,
             )
-
         if operation not in ["add", "remove", "remove_directory"]:
             return make_response(
                 jsonify(
@@ -716,14 +712,12 @@ class ManageSourceFiles(Resource):
                 ),
                 400,
             )
-
         try:
             ObjectId(source_id)
         except Exception:
             return make_response(
                 jsonify({"success": False, "message": "Invalid source ID format"}), 400
             )
-
         try:
             source = sources_collection.find_one(
                 {"_id": ObjectId(source_id), "user": user}
@@ -743,7 +737,6 @@ class ManageSourceFiles(Resource):
             return make_response(
                 jsonify({"success": False, "message": "Database error"}), 500
             )
-
         try:
             storage = StorageCreator.get_storage()
             source_file_path = source.get("file_path", "")
@@ -756,7 +749,6 @@ class ManageSourceFiles(Resource):
                     ),
                     400,
                 )
-
             if operation == "add":
                 files = request.files.getlist("file")
                 if not files or all(file.filename == "" for file in files):
@@ -769,23 +761,22 @@ class ManageSourceFiles(Resource):
                         ),
                         400,
                     )
-
                 added_files = []
 
                 target_dir = source_file_path
                 if parent_dir:
                     target_dir = f"{source_file_path}/{parent_dir}"
-
                 for file in files:
                     if file.filename:
                         safe_filename_str = safe_filename(file.filename)
                         file_path = f"{target_dir}/{safe_filename_str}"
 
                         # Save file to storage
+
                         storage.save_file(file, file_path)
                         added_files.append(safe_filename_str)
-
                 # Trigger re-ingestion pipeline
+
                 from application.api.user.tasks import reingest_source_task
 
                 task = reingest_source_task.delay(source_id=source_id, user=user)
@@ -802,7 +793,6 @@ class ManageSourceFiles(Resource):
                     ),
                     200,
                 )
-
             elif operation == "remove":
                 file_paths_str = request.form.get("file_paths")
                 if not file_paths_str:
@@ -815,7 +805,6 @@ class ManageSourceFiles(Resource):
                         ),
                         400,
                     )
-
                 try:
                     file_paths = (
                         json.loads(file_paths_str)
@@ -829,18 +818,19 @@ class ManageSourceFiles(Resource):
                         ),
                         400,
                     )
-
                 # Remove files from storage and directory structure
+
                 removed_files = []
                 for file_path in file_paths:
                     full_path = f"{source_file_path}/{file_path}"
 
                     # Remove from storage
+
                     if storage.file_exists(full_path):
                         storage.delete_file(full_path)
                         removed_files.append(file_path)
-
                 # Trigger re-ingestion pipeline
+
                 from application.api.user.tasks import reingest_source_task
 
                 task = reingest_source_task.delay(source_id=source_id, user=user)
@@ -856,7 +846,6 @@ class ManageSourceFiles(Resource):
                     ),
                     200,
                 )
-
             elif operation == "remove_directory":
                 directory_path = request.form.get("directory_path")
                 if not directory_path:
@@ -869,8 +858,8 @@ class ManageSourceFiles(Resource):
                         ),
                         400,
                     )
-
                 # Validate directory path (prevent path traversal)
+
                 if directory_path.startswith("/") or ".." in directory_path:
                     current_app.logger.warning(
                         f"Invalid directory path attempted for removal. "
@@ -882,7 +871,6 @@ class ManageSourceFiles(Resource):
                         ),
                         400,
                     )
-
                 full_directory_path = (
                     f"{source_file_path}/{directory_path}"
                     if directory_path
@@ -904,7 +892,6 @@ class ManageSourceFiles(Resource):
                         ),
                         404,
                     )
-
                 success = storage.remove_directory(full_directory_path)
 
                 if not success:
@@ -919,7 +906,6 @@ class ManageSourceFiles(Resource):
                         ),
                         500,
                     )
-
                 current_app.logger.info(
                     f"Successfully removed directory. "
                     f"User: {user}, Source ID: {source_id}, Directory path: {directory_path}, "
@@ -927,6 +913,7 @@ class ManageSourceFiles(Resource):
                 )
 
                 # Trigger re-ingestion pipeline
+
                 from application.api.user.tasks import reingest_source_task
 
                 task = reingest_source_task.delay(source_id=source_id, user=user)
@@ -942,7 +929,6 @@ class ManageSourceFiles(Resource):
                     ),
                     200,
                 )
-
         except Exception as err:
             error_context = f"operation={operation}, user={user}, source_id={source_id}"
             if operation == "remove_directory":
@@ -954,7 +940,6 @@ class ManageSourceFiles(Resource):
             elif operation == "add":
                 parent_dir = request.form.get("parent_dir", "")
                 error_context += f", parent_dir={parent_dir}"
-
             current_app.logger.error(
                 f"Error managing source files: {err} ({error_context})", exc_info=True
             )
@@ -1001,6 +986,50 @@ class UploadRemote(Resource):
                 source_data = config.get("url")
             elif data["source"] == "reddit":
                 source_data = config
+            elif data["source"] in ConnectorCreator.get_supported_connectors():
+                session_token = config.get("session_token")
+                if not session_token:
+                    return make_response(
+                        jsonify(
+                            {
+                                "success": False,
+                                "error": f"Missing session_token in {data['source']} configuration",
+                            }
+                        ),
+                        400,
+                    )
+                # Process file_ids
+
+                file_ids = config.get("file_ids", [])
+                if isinstance(file_ids, str):
+                    file_ids = [id.strip() for id in file_ids.split(",") if id.strip()]
+                elif not isinstance(file_ids, list):
+                    file_ids = []
+                # Process folder_ids
+
+                folder_ids = config.get("folder_ids", [])
+                if isinstance(folder_ids, str):
+                    folder_ids = [
+                        id.strip() for id in folder_ids.split(",") if id.strip()
+                    ]
+                elif not isinstance(folder_ids, list):
+                    folder_ids = []
+                config["file_ids"] = file_ids
+                config["folder_ids"] = folder_ids
+
+                task = ingest_connector_task.delay(
+                    job_name=data["name"],
+                    user=decoded_token.get("sub"),
+                    source_type=data["source"],
+                    session_token=session_token,
+                    file_ids=file_ids,
+                    folder_ids=folder_ids,
+                    recursive=config.get("recursive", False),
+                    retriever=config.get("retriever", "classic"),
+                )
+                return make_response(
+                    jsonify({"success": True, "task_id": task.id}), 200
+                )
             task = ingest_remote.delay(
                 source_data=source_data,
                 job_name=data["name"],
@@ -1109,6 +1138,7 @@ class PaginatedSources(Resource):
                     "retriever": doc.get("retriever", "classic"),
                     "syncFrequency": doc.get("sync_frequency", ""),
                     "isNested": bool(doc.get("directory_structure")),
+                    "type": doc.get("type", "file"),
                 }
                 paginated_docs.append(doc_data)
             response = {
@@ -1157,6 +1187,9 @@ class CombinedJson(Resource):
                         "retriever": index.get("retriever", "classic"),
                         "syncFrequency": index.get("sync_frequency", ""),
                         "is_nested": bool(index.get("directory_structure")),
+                        "type": index.get(
+                            "type", "file"
+                        ),  # Add type field with default "file"
                     }
                 )
         except Exception as err:
@@ -1372,17 +1405,14 @@ class GetAgent(Resource):
     def get(self):
         if not (decoded_token := request.decoded_token):
             return {"success": False}, 401
-
         if not (agent_id := request.args.get("id")):
             return {"success": False, "message": "ID required"}, 400
-
         try:
             agent = agents_collection.find_one(
                 {"_id": ObjectId(agent_id), "user": decoded_token["sub"]}
             )
             if not agent:
                 return {"status": "Not found"}, 404
-
             data = {
                 "id": str(agent["_id"]),
                 "name": agent["name"],
@@ -1428,7 +1458,6 @@ class GetAgent(Resource):
                 "shared_token": agent.get("shared_token", ""),
             }
             return make_response(jsonify(data), 200)
-
         except Exception as e:
             current_app.logger.error(f"Agent fetch error: {e}", exc_info=True)
             return {"success": False}, 400
@@ -1440,7 +1469,6 @@ class GetAgents(Resource):
     def get(self):
         if not (decoded_token := request.decoded_token):
             return {"success": False}, 401
-
         user = decoded_token.get("sub")
         try:
             user_doc = ensure_user_doc(user)
@@ -1501,7 +1529,6 @@ class GetAgents(Resource):
                 for agent in agents
                 if "source" in agent or "retriever" in agent
             ]
-
         except Exception as err:
             current_app.logger.error(f"Error retrieving agents: {err}", exc_info=True)
             return make_response(jsonify({"success": False}), 400)
@@ -1573,9 +1600,11 @@ class CreateAgent(Resource):
         print(f"Received data: {data}")
 
         # Validate JSON schema if provided
+
         if data.get("json_schema"):
             try:
                 # Basic validation - ensure it's a valid JSON structure
+
                 json_schema = data.get("json_schema")
                 if not isinstance(json_schema, dict):
                     return make_response(
@@ -1587,8 +1616,8 @@ class CreateAgent(Resource):
                         ),
                         400,
                     )
-
                 # Validate that it has either a 'schema' property or is itself a schema
+
                 if "schema" not in json_schema and "type" not in json_schema:
                     return make_response(
                         jsonify(
@@ -1606,7 +1635,6 @@ class CreateAgent(Resource):
                     ),
                     400,
                 )
-
         if data.get("status") not in ["draft", "published"]:
             return make_response(
                 jsonify(
@@ -1617,7 +1645,6 @@ class CreateAgent(Resource):
                 ),
                 400,
             )
-
         if data.get("status") == "published":
             required_fields = [
                 "name",
@@ -1628,6 +1655,7 @@ class CreateAgent(Resource):
                 "agent_type",
             ]
             # Require either source or sources (but not both)
+
             if not data.get("source") and not data.get("sources"):
                 return make_response(
                     jsonify(
@@ -1648,13 +1676,11 @@ class CreateAgent(Resource):
             return missing_fields
         if invalid_fields:
             return invalid_fields
-
         image_url, error = handle_image_upload(request, "", user, storage)
         if error:
             return make_response(
                 jsonify({"success": False, "message": "Image upload failed"}), 400
             )
-
         try:
             key = str(uuid.uuid4()) if data.get("status") == "published" else ""
 
@@ -1674,7 +1700,6 @@ class CreateAgent(Resource):
                     source_field = DBRef("sources", ObjectId(source_value))
                 else:
                     source_field = ""
-
             new_agent = {
                 "user": user,
                 "name": data.get("name"),
@@ -1772,7 +1797,6 @@ class UpdateAgent(Resource):
                     data["json_schema"] = json.loads(data["json_schema"])
                 except json.JSONDecodeError:
                     data["json_schema"] = None
-
         if not ObjectId.is_valid(agent_id):
             return make_response(
                 jsonify({"success": False, "message": "Invalid agent ID format"}), 400
@@ -1796,7 +1820,6 @@ class UpdateAgent(Resource):
                 ),
                 404,
             )
-
         image_url, error = handle_image_upload(
             request, existing_agent.get("image", ""), user, storage
         )
@@ -1804,7 +1827,6 @@ class UpdateAgent(Resource):
             return make_response(
                 jsonify({"success": False, "message": "Image upload failed"}), 400
             )
-
         update_fields = {}
         allowed_fields = [
             "name",
@@ -1837,6 +1859,7 @@ class UpdateAgent(Resource):
                     source_id = data.get("source")
                     if source_id == "default":
                         # Handle special "default" source
+
                         update_fields[field] = "default"
                     elif source_id and ObjectId.is_valid(source_id):
                         update_fields[field] = DBRef("sources", ObjectId(source_id))
@@ -1941,7 +1964,6 @@ class UpdateAgent(Resource):
                     ),
                     400,
                 )
-
             if not existing_agent.get("key"):
                 newly_generated_key = str(uuid.uuid4())
                 update_fields["key"] = newly_generated_key
@@ -2028,7 +2050,6 @@ class PinnedAgents(Resource):
         decoded_token = request.decoded_token
         if not decoded_token:
             return make_response(jsonify({"success": False}), 401)
-
         user_id = decoded_token.get("sub")
 
         try:
@@ -2037,7 +2058,6 @@ class PinnedAgents(Resource):
 
             if not pinned_ids:
                 return make_response(jsonify([]), 200)
-
             pinned_object_ids = [ObjectId(agent_id) for agent_id in pinned_ids]
 
             pinned_agents_cursor = agents_collection.find(
@@ -2047,6 +2067,7 @@ class PinnedAgents(Resource):
             existing_ids = {str(agent["_id"]) for agent in pinned_agents}
 
             # Clean up any stale pinned IDs
+
             stale_ids = [
                 agent_id for agent_id in pinned_ids if agent_id not in existing_ids
             ]
@@ -2055,7 +2076,6 @@ class PinnedAgents(Resource):
                     {"user_id": user_id},
                     {"$pullAll": {"agent_preferences.pinned": stale_ids}},
                 )
-
             list_pinned_agents = [
                 {
                     "id": str(agent["_id"]),
@@ -2092,11 +2112,9 @@ class PinnedAgents(Resource):
                 for agent in pinned_agents
                 if "source" in agent or "retriever" in agent
             ]
-
         except Exception as err:
             current_app.logger.error(f"Error retrieving pinned agents: {err}")
             return make_response(jsonify({"success": False}), 400)
-
         return make_response(jsonify(list_pinned_agents), 200)
 
 
@@ -2160,7 +2178,6 @@ class RemoveSharedAgent(Resource):
             return make_response(
                 jsonify({"success": False, "message": "ID is required"}), 400
             )
-
         try:
             agent = agents_collection.find_one(
                 {"_id": ObjectId(agent_id), "shared_publicly": True}
@@ -2170,7 +2187,6 @@ class RemoveSharedAgent(Resource):
                     jsonify({"success": False, "message": "Shared agent not found"}),
                     404,
                 )
-
             ensure_user_doc(user_id)
             users_collection.update_one(
                 {"user_id": user_id},
@@ -2183,7 +2199,6 @@ class RemoveSharedAgent(Resource):
             )
 
             return make_response(jsonify({"success": True, "action": "removed"}), 200)
-
         except Exception as err:
             current_app.logger.error(f"Error removing shared agent: {err}")
             return make_response(
@@ -2206,7 +2221,6 @@ class SharedAgent(Resource):
             return make_response(
                 jsonify({"success": False, "message": "Token or ID is required"}), 400
             )
-
         try:
             query = {
                 "shared_publicly": True,
@@ -2218,7 +2232,6 @@ class SharedAgent(Resource):
                     jsonify({"success": False, "message": "Shared agent not found"}),
                     404,
                 )
-
             agent_id = str(shared_agent["_id"])
             data = {
                 "id": agent_id,
@@ -2230,7 +2243,12 @@ class SharedAgent(Resource):
                     else ""
                 ),
                 "description": shared_agent.get("description", ""),
-                "source": shared_agent.get("source", ""),
+                "source": (
+                    str(source_doc["_id"])
+                    if isinstance(shared_agent.get("source"), DBRef)
+                    and (source_doc := db.dereference(shared_agent.get("source")))
+                    else ""
+                ),
                 "chunks": shared_agent.get("chunks", "0"),
                 "retriever": shared_agent.get("retriever", "classic"),
                 "prompt_id": shared_agent.get("prompt_id", "default"),
@@ -2253,7 +2271,6 @@ class SharedAgent(Resource):
                     if tool_data:
                         enriched_tools.append(tool_data.get("name", ""))
                 data["tools"] = enriched_tools
-
             decoded_token = getattr(request, "decoded_token", None)
             if decoded_token:
                 user_id = decoded_token.get("sub")
@@ -2265,9 +2282,7 @@ class SharedAgent(Resource):
                         {"user_id": user_id},
                         {"$addToSet": {"agent_preferences.shared_with_me": agent_id}},
                     )
-
             return make_response(jsonify(data), 200)
-
         except Exception as err:
             current_app.logger.error(f"Error retrieving shared agent: {err}")
             return make_response(jsonify({"success": False}), 400)
@@ -2301,7 +2316,6 @@ class SharedAgents(Resource):
                     {"user_id": user_id},
                     {"$pullAll": {"agent_preferences.shared_with_me": stale_ids}},
                 )
-
             pinned_ids = set(user_doc.get("agent_preferences", {}).get("pinned", []))
 
             list_shared_agents = [
@@ -2328,7 +2342,6 @@ class SharedAgents(Resource):
             ]
 
             return make_response(jsonify(list_shared_agents), 200)
-
         except Exception as err:
             current_app.logger.error(f"Error retrieving shared agents: {err}")
             return make_response(jsonify({"success": False}), 400)
@@ -3808,22 +3821,22 @@ class GetChunks(Resource):
                 metadata = chunk.get("metadata", {})
 
                 # Filter by path if provided
+
                 if path:
                     chunk_source = metadata.get("source", "")
                     # Check if the chunk's source matches the requested path
+
                     if not chunk_source or not chunk_source.endswith(path):
                         continue
-
                 # Filter by search term if provided
+
                 if search_term:
                     text_match = search_term in chunk.get("text", "").lower()
                     title_match = search_term in metadata.get("title", "").lower()
 
                     if not (text_match or title_match):
                         continue
-
                 filtered_chunks.append(chunk)
-
             chunks = filtered_chunks
 
             total_chunks = len(chunks)
@@ -3983,7 +3996,6 @@ class UpdateChunk(Resource):
             if metadata is None:
                 metadata = {}
             metadata["token_count"] = token_count
-
         if not ObjectId.is_valid(doc_id):
             return make_response(jsonify({"error": "Invalid doc_id"}), 400)
         doc = sources_collection.find_one({"_id": ObjectId(doc_id), "user": user})
@@ -3998,7 +4010,6 @@ class UpdateChunk(Resource):
             existing_chunk = next((c for c in chunks if c["doc_id"] == chunk_id), None)
             if not existing_chunk:
                 return make_response(jsonify({"error": "Chunk not found"}), 404)
-
             new_text = text if text is not None else existing_chunk["text"]
 
             if metadata is not None:
@@ -4006,10 +4017,8 @@ class UpdateChunk(Resource):
                 new_metadata.update(metadata)
             else:
                 new_metadata = existing_chunk["metadata"].copy()
-
             if text is not None:
                 new_metadata["token_count"] = num_tokens_from_string(new_text)
-
             try:
                 new_chunk_id = store.add_chunk(new_text, new_metadata)
 
@@ -4018,7 +4027,6 @@ class UpdateChunk(Resource):
                     current_app.logger.warning(
                         f"Failed to delete old chunk {chunk_id}, but new chunk {new_chunk_id} was created"
                     )
-
                 return make_response(
                     jsonify(
                         {
@@ -4065,7 +4073,6 @@ class StoreAttachment(Resource):
                 jsonify({"status": "error", "message": "Missing file"}),
                 400,
             )
-
         user = None
         if decoded_token:
             user = safe_filename(decoded_token.get("sub"))
@@ -4080,7 +4087,6 @@ class StoreAttachment(Resource):
             return make_response(
                 jsonify({"success": False, "message": "Authentication required"}), 401
             )
-
         try:
             attachment_id = ObjectId()
             original_filename = safe_filename(os.path.basename(file.filename))
@@ -4122,7 +4128,6 @@ class ServeImage(Resource):
             content_type = f"image/{extension}"
             if extension == "jpg":
                 content_type = "image/jpeg"
-
             response = make_response(file_obj.read())
             response.headers.set("Content-Type", content_type)
             response.headers.set("Cache-Control", "max-age=86400")
@@ -4149,36 +4154,43 @@ class DirectoryStructure(Resource):
         decoded_token = request.decoded_token
         if not decoded_token:
             return make_response(jsonify({"success": False}), 401)
-
         user = decoded_token.get("sub")
         doc_id = request.args.get("id")
 
         if not doc_id:
             return make_response(jsonify({"error": "Document ID is required"}), 400)
-
         if not ObjectId.is_valid(doc_id):
             return make_response(jsonify({"error": "Invalid document ID"}), 400)
-
         try:
             doc = sources_collection.find_one({"_id": ObjectId(doc_id), "user": user})
             if not doc:
                 return make_response(
                     jsonify({"error": "Document not found or access denied"}), 404
                 )
-
             directory_structure = doc.get("directory_structure", {})
+            base_path = doc.get("file_path", "")
 
+            provider = None
+            remote_data = doc.get("remote_data")
+            try:
+                if isinstance(remote_data, str) and remote_data:
+                    remote_data_obj = json.loads(remote_data)
+                    provider = remote_data_obj.get("provider")
+            except Exception as e:
+                current_app.logger.warning(
+                    f"Failed to parse remote_data for doc {doc_id}: {e}"
+                )
             return make_response(
                 jsonify(
                     {
                         "success": True,
                         "directory_structure": directory_structure,
-                        "base_path": doc.get("file_path", ""),
+                        "base_path": base_path,
+                        "provider": provider,
                     }
                 ),
                 200,
             )
-
         except Exception as e:
             current_app.logger.error(
                 f"Error retrieving directory structure: {e}", exc_info=True

application/api/user/tasks.py

@@ -47,6 +47,39 @@ def process_agent_webhook(self, agent_id, payload):
     return resp
 
 
+@celery.task(bind=True)
+def ingest_connector_task(
+    self,
+    job_name,
+    user,
+    source_type,
+    session_token=None,
+    file_ids=None,
+    folder_ids=None,
+    recursive=True,
+    retriever="classic",
+    operation_mode="upload",
+    doc_id=None,
+    sync_frequency="never"
+):
+    from application.worker import ingest_connector
+    resp = ingest_connector(
+        self,
+        job_name,
+        user,
+        source_type,
+        session_token=session_token,
+        file_ids=file_ids,
+        folder_ids=folder_ids,
+        recursive=recursive,
+        retriever=retriever,
+        operation_mode=operation_mode,
+        doc_id=doc_id,
+        sync_frequency=sync_frequency
+    )
+    return resp
+
+
 @celery.on_after_configure.connect
 def setup_periodic_tasks(sender, **kwargs):
     sender.add_periodic_task(