feat: remote MCP server integration with admin UI and Connector Manager gateway

connectors/atlassian/src/sync.rs

@@ -101,7 +101,8 @@ impl SyncManager {
             return Err(anyhow!("Source is not active: {}", source_id));
         }
 
-        let source_type = source.source_type;
+        let source_type = SourceType::try_from(source.source_type.as_str())
+            .map_err(|e| anyhow!("Invalid source type for Atlassian connector: {}", e))?;
         if source_type != SourceType::Confluence && source_type != SourceType::Jira {
             return Err(anyhow!(
                 "Invalid source type for Atlassian connector: {:?}",

connectors/atlassian/tests/common/mock_atlassian.rs

@@ -5,15 +5,15 @@ use std::collections::HashMap;
 use std::pin::Pin;
 use std::sync::Mutex;
 
-use omni_atlassian_connector::AtlassianApi;
-use omni_atlassian_connector::AtlassianCredentials;
 use omni_atlassian_connector::client::{OrgGroupInfo, PageReadRestrictions};
 use omni_atlassian_connector::models::{
     ConfluenceCqlPage, ConfluencePage, ConfluenceSpace, ConfluenceSpacePermission, JiraField,
     JiraIssue, JiraIssueSecuritySchemeResponse, JiraPermissionSchemeResponse,
     JiraProjectIssueSecuritySchemeResponse, JiraProjectRolesResponse, JiraRoleActorsResponse,
     JiraSearchResponse, JiraSecurityLevelMember,
 };
+use omni_atlassian_connector::AtlassianApi;
+use omni_atlassian_connector::AtlassianCredentials;
 
 #[derive(Debug, Clone)]
 pub struct MethodCall {

connectors/atlassian/tests/common/mod.rs

@@ -7,7 +7,8 @@ use omni_connector_sdk::SdkClient;
 use redis::AsyncCommands;
 use shared::db::repositories::service_credentials::ServiceCredentialsRepo;
 use shared::models::{
-    AuthType, ConnectorManifest, ServiceCredential, ServiceProvider, SourceType, SyncType,
+    AuthType, ConnectorManifest, IntegrationType, ServiceCredential, ServiceProvider, SourceType,
+    SyncType,
 };
 use shared::storage::postgres::PostgresStorage;
 use shared::test_environment::TestEnvironment;
@@ -81,7 +82,11 @@ pub async fn setup_test_fixture(source_type: SourceType) -> Result<TestFixture>
         sync_modes: vec![SyncType::Full],
         connector_id: "atlassian".to_string(),
         connector_url: "http://127.0.0.1:1".to_string(),
-        source_types: vec![SourceType::Confluence, SourceType::Jira],
+        integration_type: IntegrationType::Connector,
+        source_types: vec![
+            SourceType::Confluence.to_string(),
+            SourceType::Jira.to_string(),
+        ],
         description: None,
         actions: vec![],
         search_operators: vec![],

connectors/google/src/connector.rs

@@ -1197,7 +1197,9 @@ impl Connector for GoogleConnector {
                 }
             }
             _ => {
-                create_service_auth(creds, source.source_type).map_err(|e| {
+                let native_source_type = SourceType::try_from(source.source_type.as_str())
+                    .map_err(SyncRequestValidationError::BadRequest)?;
+                create_service_auth(creds, native_source_type).map_err(|e| {
                     SyncRequestValidationError::BadRequest(format!(
                         "Invalid Google service-account credentials: {}",
                         e

connectors/google/src/sync.rs

@@ -991,7 +991,10 @@ impl SyncManager {
         // mid-sync.
         let existing_state = existing_state.unwrap_or_default();
 
-        let result = match source.source_type {
+        let native_source_type = SourceType::try_from(source.source_type.as_str())
+            .map_err(|e| anyhow!("Unsupported source type: {}", e))?;
+
+        let result = match native_source_type {
             SourceType::GoogleDrive => {
                 self.sync_drive_source_internal(
                     source,
@@ -1027,7 +1030,7 @@ impl SyncManager {
             _ => Err(anyhow!("Unsupported source type: {:?}", source.source_type)),
         };
 
-        if result.is_ok() && source.source_type == SourceType::GoogleDrive {
+        if result.is_ok() && native_source_type == SourceType::GoogleDrive {
             self.ensure_webhook_registered(source_id).await;
         }
 
@@ -1851,7 +1854,9 @@ impl SyncManager {
         ctx: &SyncContext,
     ) -> Result<GoogleSyncCheckpoint> {
         let sync_run_id = ctx.sync_run_id();
-        let service_auth = Arc::new(self.create_auth(service_creds, source.source_type).await?);
+        let native_source_type = SourceType::try_from(source.source_type.as_str())
+            .map_err(|e| anyhow!("Unsupported source type: {}", e))?;
+        let service_auth = Arc::new(self.create_auth(service_creds, native_source_type).await?);
         let (drive_cutoff_date, _gmail_cutoff_date) = self.get_cutoff_date()?;
         let drive_cutoff = parse_google_time(Some(&drive_cutoff_date))
             .ok_or_else(|| anyhow!("Invalid Drive cutoff time: {}", drive_cutoff_date))?;
@@ -2289,7 +2294,10 @@ impl SyncManager {
         ctx: &SyncContext,
     ) -> Result<GoogleSyncCheckpoint> {
         let sync_run_id = ctx.sync_run_id();
-        let service_auth = Arc::new(self.create_auth(service_creds, source.source_type).await?);
+
+        let native_source_type = SourceType::try_from(source.source_type.as_str())
+            .map_err(|e| anyhow!("Unsupported source type: {}", e))?;
+        let service_auth = Arc::new(self.create_auth(service_creds, native_source_type).await?);
 
         // Parse folder-path filters. Malformed config is a sync error.
         let parsed_filters = match crate::models::parse_folder_path_filters(&source.config) {
@@ -2612,7 +2620,9 @@ impl SyncManager {
     ) -> Result<GoogleSyncCheckpoint> {
         let sync_run_id = ctx.sync_run_id();
 
-        let service_auth = Arc::new(self.create_auth(service_creds, source.source_type).await?);
+        let native_source_type = SourceType::try_from(source.source_type.as_str())
+            .map_err(|e| anyhow!("Unsupported source type: {}", e))?;
+        let service_auth = Arc::new(self.create_auth(service_creds, native_source_type).await?);
 
         let (_drive_cutoff_date, gmail_cutoff_date) = self.get_cutoff_date()?;
         info!("Using Gmail cutoff date: {}", gmail_cutoff_date);
@@ -4911,7 +4921,14 @@ impl SyncManager {
         service_creds: &ServiceCredential,
         ctx: &SyncContext,
     ) -> HashSet<String> {
-        let service_auth = match self.create_auth(service_creds, source.source_type).await {
+        let native_source_type = match SourceType::try_from(source.source_type.as_str()) {
+            Ok(source_type) => source_type,
+            Err(e) => {
+                warn!("Skipping group sync for unsupported source type: {}", e);
+                return HashSet::new();
+            }
+        };
+        let service_auth = match self.create_auth(service_creds, native_source_type).await {
             Ok(auth) => auth,
             Err(e) => {
                 warn!("Failed to create auth for group sync: {}", e);

sdk/rust/src/connector.rs

@@ -13,7 +13,7 @@ use serde::de::DeserializeOwned;
 use serde::Serialize;
 use serde_json::Value as JsonValue;
 use shared::models::{
-    ActionDefinition, ConnectorManifest, ConnectorSkillDefinition, SearchOperator,
+    ActionDefinition, ConnectorManifest, ConnectorSkillDefinition, IntegrationType, SearchOperator,
     ServiceCredential, Source, SourceType, SyncType,
 };
 
@@ -165,7 +165,12 @@ pub trait Connector: Send + Sync + 'static {
             sync_modes: self.sync_modes(),
             connector_id: self.name().to_string(),
             connector_url,
-            source_types: self.source_types(),
+            integration_type: IntegrationType::Connector,
+            source_types: self
+                .source_types()
+                .into_iter()
+                .map(|source_type| source_type.to_string())
+                .collect(),
             description: self.description(),
             actions: self.actions(),
             search_operators: self.search_operators(),

sdk/rust/src/lib.rs

@@ -20,9 +20,9 @@ pub use server::{create_router, serve, serve_with_config, serve_with_extra_route
 pub use shared::models::DocumentAttributes;
 pub use shared::models::{
     ActionDefinition, ActionMode, AuthType, ConnectorEvent, ConnectorManifest,
-    ConnectorSkillDefinition, DocumentMetadata, DocumentPermissions, McpPromptDefinition,
-    McpResourceDefinition, SearchOperator, ServiceCredential, ServiceProvider, Source, SourceType,
-    SyncRun, SyncStatus, SyncType,
+    ConnectorSkillDefinition, DocumentMetadata, DocumentPermissions, IntegrationType,
+    McpPromptDefinition, McpResourceDefinition, SearchOperator, ServiceCredential, ServiceProvider,
+    Source, SourceType, SyncRun, SyncStatus, SyncType,
 };
 pub use shared::rate_limiter::{RateLimiter, RetryableError};
 pub use shared::telemetry;

sdk/rust/src/server.rs

@@ -18,7 +18,7 @@ use axum::{
 use dashmap::mapref::entry::Entry;
 use dashmap::DashMap;
 use serde::de::DeserializeOwned;
-use shared::models::{ConnectorSkillDefinition, SyncSlotClass, SyncType};
+use shared::models::{ConnectorSkillDefinition, SourceType, SyncSlotClass, SyncType};
 use shared::telemetry;
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -482,11 +482,21 @@ where
         .register_sync(&sync_run_id, request.sync_mode)
         .await;
 
+    let native_source_type = SourceType::try_from(source.source_type.as_str()).map_err(|e| {
+        (
+            StatusCode::BAD_REQUEST,
+            Json(SyncResponse::error(format!(
+                "source {} is not a native connector source: {}",
+                source_id, e
+            ))),
+        )
+    })?;
+
     let ctx = SyncContext::new_with_resume(
         state.sdk_client.clone(),
         sync_run_id.clone(),
         source_id.clone(),
-        source.source_type,
+        native_source_type,
         request.sync_mode,
         request.is_resume,
         cancelled,

services/ai/agents/executor.py

@@ -8,7 +8,6 @@
 from pathlib import Path
 from typing import cast
 
-import httpx
 from anthropic.types import (
     BashCodeExecutionToolResultBlockParam,
     CodeExecutionToolResultBlockParam,
@@ -68,7 +67,7 @@
     ConnectorToolHandler,
     SourceFilter,
     ToolsetSummary,
-    sources_from_sync_overview_response,
+    fetch_active_sources_from_connector_manager,
 )
 from tools.email_handler import EmailToolHandler
 from tools.mcp_capability_handler import McpCapabilityHandler
@@ -125,12 +124,9 @@ async def _resolve_llm_provider(state: AppState, agent: Agent) -> ResolvedModel:
 
 
 async def _fetch_sources() -> list[Source] | None:
-    """Fetch all sources from the connector manager."""
+    """Fetch active sources, including remote MCP rows, from connector manager."""
     try:
-        async with httpx.AsyncClient(timeout=10.0) as client:
-            resp = await client.get(f"{CONNECTOR_MANAGER_URL.rstrip('/')}/sources")
-            resp.raise_for_status()
-            return sources_from_sync_overview_response(resp.json())
+        return await fetch_active_sources_from_connector_manager(CONNECTOR_MANAGER_URL)
     except Exception as e:
         logger.warning(f"Failed to fetch sources: {e}")
         return None

services/ai/db/models.py

@@ -415,6 +415,7 @@ class Source:
     source_type: str
     is_active: bool
     is_deleted: bool
+    integration_type: str = "connector"
 
     @classmethod
     def from_row(cls, row: Mapping[str, object]) -> "Source":
@@ -423,6 +424,7 @@ def from_row(cls, row: Mapping[str, object]) -> "Source":
             name=cast(str, row["name"]),
             source_type=cast(str, row["source_type"]),
             is_active=cast(bool, row["is_active"]),
+            integration_type=cast(str, row.get("integration_type", "connector")),
             is_deleted=cast(bool, row["is_deleted"]),
         )
 

services/ai/routers/chat.py

@@ -102,7 +102,7 @@
 from tools.connector_handler import (
     SearchOperator,
     ToolsetSummary,
-    sources_from_sync_overview_response,
+    fetch_active_sources_from_connector_manager,
 )
 from tools.mcp_capability_handler import McpCapabilityHandler
 from tools.meta_handler import MetaToolHandler, OnLoad
@@ -233,10 +233,7 @@ def _loaded_source_ids(
 
 async def _fetch_sources_from_connector_manager() -> list[Source] | None:
     try:
-        async with httpx.AsyncClient(timeout=10.0) as client:
-            resp = await client.get(f"{CONNECTOR_MANAGER_URL.rstrip('/')}/sources")
-            resp.raise_for_status()
-            return sources_from_sync_overview_response(resp.json())
+        return await fetch_active_sources_from_connector_manager(CONNECTOR_MANAGER_URL)
     except Exception as e:
         logger.warning(f"Failed to fetch sources from connector manager: {e}")
         return None