Merge branch 'main' into feat/remote-mcp

application/api/answer/services/stream_processor.py

@@ -69,11 +69,8 @@ class StreamProcessor:
             self.decoded_token.get("sub") if self.decoded_token is not None else None
         )
         self.conversation_id = self.data.get("conversation_id")
-        self.source = (
-            {"active_docs": self.data["active_docs"]}
-            if "active_docs" in self.data
-            else {}
-        )
+        self.source = {}
+        self.all_sources = []
         self.attachments = []
         self.history = []
         self.agent_config = {}
@@ -85,6 +82,8 @@ class StreamProcessor:
 
     def initialize(self):
         """Initialize all required components for processing"""
+        self._configure_agent()
+        self._configure_source()
         self._configure_retriever()
         self._configure_agent()
         self._load_conversation_history()
@@ -171,13 +170,77 @@ class StreamProcessor:
         source = data.get("source")
         if isinstance(source, DBRef):
             source_doc = self.db.dereference(source)
-            data["source"] = str(source_doc["_id"])
-            data["retriever"] = source_doc.get("retriever", data.get("retriever"))
-            data["chunks"] = source_doc.get("chunks", data.get("chunks"))
+            if source_doc:
+                data["source"] = str(source_doc["_id"])
+                data["retriever"] = source_doc.get("retriever", data.get("retriever"))
+                data["chunks"] = source_doc.get("chunks", data.get("chunks"))
+            else:
+                data["source"] = None
+        elif source == "default":
+            data["source"] = "default"
         else:
             data["source"] = None
+        # Handle multiple sources
+
+        sources = data.get("sources", [])
+        if sources and isinstance(sources, list):
+            sources_list = []
+            for i, source_ref in enumerate(sources):
+                if source_ref == "default":
+                    processed_source = {
+                        "id": "default",
+                        "retriever": "classic",
+                        "chunks": data.get("chunks", "2"),
+                    }
+                    sources_list.append(processed_source)
+                elif isinstance(source_ref, DBRef):
+                    source_doc = self.db.dereference(source_ref)
+                    if source_doc:
+                        processed_source = {
+                            "id": str(source_doc["_id"]),
+                            "retriever": source_doc.get("retriever", "classic"),
+                            "chunks": source_doc.get("chunks", data.get("chunks", "2")),
+                        }
+                        sources_list.append(processed_source)
+            data["sources"] = sources_list
+        else:
+            data["sources"] = []
         return data
 
+    def _configure_source(self):
+        """Configure the source based on agent data"""
+        api_key = self.data.get("api_key") or self.agent_key
+
+        if api_key:
+            agent_data = self._get_data_from_api_key(api_key)
+
+            if agent_data.get("sources") and len(agent_data["sources"]) > 0:
+                source_ids = [
+                    source["id"] for source in agent_data["sources"] if source.get("id")
+                ]
+                if source_ids:
+                    self.source = {"active_docs": source_ids}
+                else:
+                    self.source = {}
+                self.all_sources = agent_data["sources"]
+            elif agent_data.get("source"):
+                self.source = {"active_docs": agent_data["source"]}
+                self.all_sources = [
+                    {
+                        "id": agent_data["source"],
+                        "retriever": agent_data.get("retriever", "classic"),
+                    }
+                ]
+            else:
+                self.source = {}
+                self.all_sources = []
+            return
+        if "active_docs" in self.data:
+            self.source = {"active_docs": self.data["active_docs"]}
+            return
+        self.source = {}
+        self.all_sources = []
+
     def _configure_agent(self):
         """Configure the agent based on request data"""
         agent_id = self.data.get("agent_id")
@@ -203,7 +266,13 @@ class StreamProcessor:
             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"]
+                try:
+                    self.retriever_config["chunks"] = int(data_key["chunks"])
+                except (ValueError, TypeError):
+                    logger.warning(
+                        f"Invalid chunks value: {data_key['chunks']}, using default value 2"
+                    )
+                    self.retriever_config["chunks"] = 2
         elif self.agent_key:
             data_key = self._get_data_from_api_key(self.agent_key)
             self.agent_config.update(
@@ -224,7 +293,13 @@ class StreamProcessor:
             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"]
+                try:
+                    self.retriever_config["chunks"] = int(data_key["chunks"])
+                except (ValueError, TypeError):
+                    logger.warning(
+                        f"Invalid chunks value: {data_key['chunks']}, using default value 2"
+                    )
+                    self.retriever_config["chunks"] = 2
         else:
             self.agent_config.update(
                 {
@@ -243,7 +318,8 @@ class StreamProcessor:
             "token_limit": self.data.get("token_limit", settings.DEFAULT_MAX_HISTORY),
         }
 
-        if "isNoneDoc" in self.data and self.data["isNoneDoc"]:
+        api_key = self.data.get("api_key") or self.agent_key
+        if not api_key and "isNoneDoc" in self.data and self.data["isNoneDoc"]:
             self.retriever_config["chunks"] = 0
 
     def create_agent(self):

application/api/connector/routes.py

@@ -0,0 +1,626 @@
+import datetime
+import json
+
+
+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

@@ -32,13 +32,15 @@ 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.security.encryption import encrypt_credentials, decrypt_credentials
+from application.parser.connectors.connector_creator import ConnectorCreator
+from application.security.encryption import decrypt_credentials, encrypt_credentials
 from application.storage.storage_creator import StorageCreator
 from application.tts.google_tts import GoogleTTS
 from application.utils import (
@@ -76,7 +78,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)
@@ -129,11 +130,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
 
 
@@ -185,7 +184,6 @@ def handle_image_upload(
                     jsonify({"success": False, "message": "Image upload failed"}),
                     400,
                 )
-
     return image_url, None
 
 
@@ -299,8 +297,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"]:
@@ -501,6 +499,7 @@ class DeleteOldIndexes(Resource):
 
         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"):
@@ -571,6 +570,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}"
@@ -592,6 +592,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 (
@@ -614,11 +615,13 @@ 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)
@@ -709,7 +712,6 @@ class ManageSourceFiles(Resource):
                 ),
                 400,
             )
-
         if operation not in ["add", "remove", "remove_directory"]:
             return make_response(
                 jsonify(
@@ -720,14 +722,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}
@@ -760,7 +760,6 @@ class ManageSourceFiles(Resource):
                     ),
                     400,
                 )
-
             if operation == "add":
                 files = request.files.getlist("file")
                 if not files or all(file.filename == "" for file in files):
@@ -773,23 +772,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)
@@ -819,7 +817,6 @@ class ManageSourceFiles(Resource):
                         ),
                         400,
                     )
-
                 try:
                     file_paths = (
                         json.loads(file_paths_str)
@@ -833,18 +830,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)
@@ -873,8 +871,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. "
@@ -908,7 +906,6 @@ class ManageSourceFiles(Resource):
                         ),
                         404,
                     )
-
                 success = storage.remove_directory(full_directory_path)
 
                 if not success:
@@ -923,7 +920,6 @@ class ManageSourceFiles(Resource):
                         ),
                         500,
                     )
-
                 current_app.logger.info(
                     f"Successfully removed directory. "
                     f"User: {user}, Source ID: {source_id}, Directory path: {directory_path}, "
@@ -931,6 +927,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)
@@ -1005,6 +1002,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"],
@@ -1113,6 +1154,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 = {
@@ -1161,6 +1203,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:
@@ -1376,17 +1421,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"],
@@ -1400,6 +1442,16 @@ class GetAgent(Resource):
                     and (source_doc := db.dereference(agent.get("source")))
                     else ""
                 ),
+                "sources": [
+                    (
+                        str(db.dereference(source_ref)["_id"])
+                        if isinstance(source_ref, DBRef) and db.dereference(source_ref)
+                        else source_ref
+                    )
+                    for source_ref in agent.get("sources", [])
+                    if (isinstance(source_ref, DBRef) and db.dereference(source_ref))
+                    or source_ref == "default"
+                ],
                 "chunks": agent["chunks"],
                 "retriever": agent.get("retriever", ""),
                 "prompt_id": agent.get("prompt_id", ""),
@@ -1422,7 +1474,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
@@ -1434,7 +1485,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)
@@ -1453,8 +1503,24 @@ class GetAgents(Resource):
                         str(source_doc["_id"])
                         if isinstance(agent.get("source"), DBRef)
                         and (source_doc := db.dereference(agent.get("source")))
-                        else ""
+                        else (
+                            agent.get("source", "")
+                            if agent.get("source") == "default"
+                            else ""
+                        )
                     ),
+                    "sources": [
+                        (
+                            source_ref
+                            if source_ref == "default"
+                            else str(db.dereference(source_ref)["_id"])
+                        )
+                        for source_ref in agent.get("sources", [])
+                        if source_ref == "default"
+                        or (
+                            isinstance(source_ref, DBRef) and db.dereference(source_ref)
+                        )
+                    ],
                     "chunks": agent["chunks"],
                     "retriever": agent.get("retriever", ""),
                     "prompt_id": agent.get("prompt_id", ""),
@@ -1497,7 +1563,14 @@ class CreateAgent(Resource):
             "image": fields.Raw(
                 required=False, description="Image file upload", type="file"
             ),
-            "source": fields.String(required=True, description="Source ID"),
+            "source": fields.String(
+                required=False, description="Source ID (legacy single source)"
+            ),
+            "sources": fields.List(
+                fields.String,
+                required=False,
+                description="List of source identifiers for multiple sources",
+            ),
             "chunks": fields.Integer(required=True, description="Chunks count"),
             "retriever": fields.String(required=True, description="Retriever ID"),
             "prompt_id": fields.String(required=True, description="Prompt ID"),
@@ -1530,6 +1603,11 @@ class CreateAgent(Resource):
                     data["tools"] = json.loads(data["tools"])
                 except json.JSONDecodeError:
                     data["tools"] = []
+            if "sources" in data:
+                try:
+                    data["sources"] = json.loads(data["sources"])
+                except json.JSONDecodeError:
+                    data["sources"] = []
             if "json_schema" in data:
                 try:
                     data["json_schema"] = json.loads(data["json_schema"])
@@ -1538,9 +1616,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(
@@ -1554,6 +1634,7 @@ class CreateAgent(Resource):
                     )
 
                 # 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(
@@ -1571,7 +1652,6 @@ class CreateAgent(Resource):
                     ),
                     400,
                 )
-
         if data.get("status") not in ["draft", "published"]:
             return make_response(
                 jsonify(
@@ -1582,17 +1662,27 @@ class CreateAgent(Resource):
                 ),
                 400,
             )
-
         if data.get("status") == "published":
             required_fields = [
                 "name",
                 "description",
-                "source",
                 "chunks",
                 "retriever",
                 "prompt_id",
                 "agent_type",
             ]
+            # Require either source or sources (but not both)
+
+            if not data.get("source") and not data.get("sources"):
+                return make_response(
+                    jsonify(
+                        {
+                            "success": False,
+                            "message": "Either 'source' or 'sources' field is required for published agents",
+                        }
+                    ),
+                    400,
+                )
             validate_fields = ["name", "description", "prompt_id", "agent_type"]
         else:
             required_fields = ["name"]
@@ -1603,25 +1693,37 @@ 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 ""
+
+            sources_list = []
+            if data.get("sources") and len(data.get("sources", [])) > 0:
+                for source_id in data.get("sources", []):
+                    if source_id == "default":
+                        sources_list.append("default")
+                    elif ObjectId.is_valid(source_id):
+                        sources_list.append(DBRef("sources", ObjectId(source_id)))
+                source_field = ""
+            else:
+                source_value = data.get("source", "")
+                if source_value == "default":
+                    source_field = "default"
+                elif ObjectId.is_valid(source_value):
+                    source_field = DBRef("sources", ObjectId(source_value))
+                else:
+                    source_field = ""
             new_agent = {
                 "user": user,
                 "name": data.get("name"),
                 "description": data.get("description", ""),
                 "image": image_url,
-                "source": (
-                    DBRef("sources", ObjectId(data.get("source")))
-                    if ObjectId.is_valid(data.get("source"))
-                    else ""
-                ),
+                "source": source_field,
+                "sources": sources_list,
                 "chunks": data.get("chunks", ""),
                 "retriever": data.get("retriever", ""),
                 "prompt_id": data.get("prompt_id", ""),
@@ -1636,7 +1738,11 @@ class CreateAgent(Resource):
             }
             if new_agent["chunks"] == "":
                 new_agent["chunks"] = "0"
-            if new_agent["source"] == "" and new_agent["retriever"] == "":
+            if (
+                new_agent["source"] == ""
+                and new_agent["retriever"] == ""
+                and not new_agent["sources"]
+            ):
                 new_agent["retriever"] = "classic"
             resp = agents_collection.insert_one(new_agent)
             new_id = str(resp.inserted_id)
@@ -1658,7 +1764,14 @@ class UpdateAgent(Resource):
             "image": fields.String(
                 required=False, description="New image URL or identifier"
             ),
-            "source": fields.String(required=True, description="Source ID"),
+            "source": fields.String(
+                required=False, description="Source ID (legacy single source)"
+            ),
+            "sources": fields.List(
+                fields.String,
+                required=False,
+                description="List of source identifiers for multiple sources",
+            ),
             "chunks": fields.Integer(required=True, description="Chunks count"),
             "retriever": fields.String(required=True, description="Retriever ID"),
             "prompt_id": fields.String(required=True, description="Prompt ID"),
@@ -1691,12 +1804,16 @@ class UpdateAgent(Resource):
                     data["tools"] = json.loads(data["tools"])
                 except json.JSONDecodeError:
                     data["tools"] = []
+            if "sources" in data:
+                try:
+                    data["sources"] = json.loads(data["sources"])
+                except json.JSONDecodeError:
+                    data["sources"] = []
             if "json_schema" in data:
                 try:
                     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
@@ -1720,7 +1837,6 @@ class UpdateAgent(Resource):
                 ),
                 404,
             )
-
         image_url, error = handle_image_upload(
             request, existing_agent.get("image", ""), user, storage
         )
@@ -1728,13 +1844,13 @@ class UpdateAgent(Resource):
             return make_response(
                 jsonify({"success": False, "message": "Image upload failed"}), 400
             )
-
         update_fields = {}
         allowed_fields = [
             "name",
             "description",
             "image",
             "source",
+            "sources",
             "chunks",
             "retriever",
             "prompt_id",
@@ -1758,7 +1874,11 @@ class UpdateAgent(Resource):
                     update_fields[field] = new_status
                 elif field == "source":
                     source_id = data.get("source")
-                    if source_id and ObjectId.is_valid(source_id):
+                    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))
                     elif source_id:
                         return make_response(
@@ -1772,6 +1892,30 @@ class UpdateAgent(Resource):
                         )
                     else:
                         update_fields[field] = ""
+                elif field == "sources":
+                    sources_list = data.get("sources", [])
+                    if sources_list and isinstance(sources_list, list):
+                        valid_sources = []
+                        for source_id in sources_list:
+                            if source_id == "default":
+                                valid_sources.append("default")
+                            elif ObjectId.is_valid(source_id):
+                                valid_sources.append(
+                                    DBRef("sources", ObjectId(source_id))
+                                )
+                            else:
+                                return make_response(
+                                    jsonify(
+                                        {
+                                            "success": False,
+                                            "message": f"Invalid source ID format: {source_id}",
+                                        }
+                                    ),
+                                    400,
+                                )
+                        update_fields[field] = valid_sources
+                    else:
+                        update_fields[field] = []
                 elif field == "chunks":
                     chunks_value = data.get("chunks")
                     if chunks_value == "":
@@ -1837,7 +1981,6 @@ class UpdateAgent(Resource):
                     ),
                     400,
                 )
-
             if not existing_agent.get("key"):
                 newly_generated_key = str(uuid.uuid4())
                 update_fields["key"] = newly_generated_key
@@ -1924,7 +2067,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:
@@ -1933,7 +2075,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(
@@ -1943,6 +2084,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
             ]
@@ -1951,7 +2093,6 @@ class PinnedAgents(Resource):
                     {"user_id": user_id},
                     {"$pullAll": {"agent_preferences.pinned": stale_ids}},
                 )
-
             list_pinned_agents = [
                 {
                     "id": str(agent["_id"]),
@@ -1988,11 +2129,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)
 
 
@@ -2056,7 +2195,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}
@@ -2066,7 +2204,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},
@@ -2079,7 +2216,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(
@@ -2102,7 +2238,6 @@ class SharedAgent(Resource):
             return make_response(
                 jsonify({"success": False, "message": "Token or ID is required"}), 400
             )
-
         try:
             query = {
                 "shared_publicly": True,
@@ -2114,7 +2249,6 @@ class SharedAgent(Resource):
                     jsonify({"success": False, "message": "Shared agent not found"}),
                     404,
                 )
-
             agent_id = str(shared_agent["_id"])
             data = {
                 "id": agent_id,
@@ -2154,7 +2288,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")
@@ -2166,9 +2299,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)
@@ -2202,7 +2333,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 = [
@@ -2229,7 +2359,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)
@@ -3762,20 +3891,21 @@ 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
@@ -3937,7 +4067,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})
@@ -3952,7 +4081,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:
@@ -3960,10 +4088,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)
 
@@ -4019,7 +4145,6 @@ class StoreAttachment(Resource):
                 jsonify({"status": "error", "message": "Missing file"}),
                 400,
             )
-
         user = None
         if decoded_token:
             user = safe_filename(decoded_token.get("sub"))
@@ -4034,7 +4159,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))
@@ -4076,7 +4200,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")
@@ -4121,18 +4244,29 @@ class DirectoryStructure(Resource):
                 )
 
             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(