diff --git a/ai-services/Makefile b/ai-services/Makefile
index c7bf4a184..96e582a86 100644
--- a/ai-services/Makefile
+++ b/ai-services/Makefile
@@ -1,6 +1,6 @@
 REGISTRY?=icr.io/ai-services-private
 IMAGE=ai-services
-TAG?=v0.0.272
+TAG?=v0.0.273
 CONTAINER_BUILDER?=podman
 CREDS_ARG := $(if $(and $(REGISTRY_USER),$(REGISTRY_PASSWORD)),--creds="$(REGISTRY_USER):$(REGISTRY_PASSWORD)")
 
diff --git a/ai-services/assets/catalog/openshift/values.yaml b/ai-services/assets/catalog/openshift/values.yaml
index de47b1903..fa365d7fb 100644
--- a/ai-services/assets/catalog/openshift/values.yaml
+++ b/ai-services/assets/catalog/openshift/values.yaml
@@ -9,7 +9,7 @@ ui:
       cpu: "500m"
 
 backend:
-  image: icr.io/ai-services-cicd/ai-services:v0.0.272
+  image: icr.io/ai-services-cicd/ai-services:v0.0.273
   runtime: "openshift"
   adminPasswordHash: ""
   # @generate:password length=32, special=false
diff --git a/ai-services/assets/catalog/podman/values.yaml b/ai-services/assets/catalog/podman/values.yaml
index ae0589ac9..c242bbc6b 100644
--- a/ai-services/assets/catalog/podman/values.yaml
+++ b/ai-services/assets/catalog/podman/values.yaml
@@ -4,7 +4,7 @@ ui:
 
 backend:
   port: ""
-  image: icr.io/ai-services-cicd/ai-services:v0.0.272
+  image: icr.io/ai-services-cicd/ai-services:v0.0.273
   runtime: ""
   adminPasswordHash: ""
   # @generate:password length=32, special=false
diff --git a/ai-services/assets/connectors/datasource/file_system/schema.json b/ai-services/assets/connectors/datasource/file_system/schema.json
index a96cf7586..cc706144d 100644
--- a/ai-services/assets/connectors/datasource/file_system/schema.json
+++ b/ai-services/assets/connectors/datasource/file_system/schema.json
@@ -14,26 +14,30 @@
       "type": "string",
       "title": "Hostname or IP address",
       "description": "Hostname or IP address of the remote file system server.",
+      "minLength": 1,
       "ui:section": "Location"
     },
     "remote_path": {
       "type": "string",
       "title": "Remote folder path",
       "description": "Absolute path on the remote file system server to use as the content root (e.g. /data/documents).",
+      "minLength": 1,
       "ui:section": "Location"
     },
     "username": {
       "type": "string",
       "title": "Username",
       "description": "Username used to authenticate.",
+      "minLength": 1,
       "ui:section": "Authentication"
     },
     "private_key": {
       "type": "string",
       "title": "Private key",
       "description": "PEM-encoded private key used for SSH authentication.",
-      "ui:section": "Authentication",
-      "format": "password"
+      "format": "password",
+      "minLength": 1,
+      "ui:section": "Authentication"
     },
     "allowed_extensions": {
       "type": "array",
diff --git a/ai-services/assets/connectors/datasource/object_storage/schema.json b/ai-services/assets/connectors/datasource/object_storage/schema.json
index 25d25fb9f..836a4c872 100644
--- a/ai-services/assets/connectors/datasource/object_storage/schema.json
+++ b/ai-services/assets/connectors/datasource/object_storage/schema.json
@@ -21,6 +21,7 @@
       "type": "string",
       "title": "Bucket name",
       "description": "Name of the S3 bucket to use as the content source.",
+      "minLength": 1,
       "ui:section": "Location"
     },
     "prefix": {
@@ -39,6 +40,7 @@
       "type": "string",
       "title": "Access key ID",
       "description": "S3-compatible access key identifier.",
+      "minLength": 1,
       "ui:section": "Authentication"
     },
     "secret_access_key": {
@@ -46,6 +48,7 @@
       "title": "Secret access key",
       "description": "S3-compatible secret access key.",
       "format": "password",
+      "minLength": 1,
       "ui:section": "Authentication"
     },
     "allowed_extensions": {
diff --git a/ai-services/cmd/ai-services/cmd/catalog/apiserver.go b/ai-services/cmd/ai-services/cmd/catalog/apiserver.go
index edaf5a956..42a2405b7 100644
--- a/ai-services/cmd/ai-services/cmd/catalog/apiserver.go
+++ b/ai-services/cmd/ai-services/cmd/catalog/apiserver.go
@@ -83,6 +83,7 @@ func buildAPIServerOptions(ctx context.Context, pool *pgxpool.Pool, secretKey, a
 	svcRepo := repository.NewServiceRepository(pool)
 	compRepo := repository.NewComponentRepository(pool)
 	svcDepRepo := repository.NewServiceDependencyRepository(pool)
+	connectorRepo := repository.NewConnectorRepository(pool)
 
 	catalogProvider, err := catalog.NewCatalogProvider(bundleRepo)
 	if err != nil {
@@ -97,6 +98,13 @@ func buildAPIServerOptions(ctx context.Context, pool *pgxpool.Pool, secretKey, a
 	}
 	syncService.Start(ctx)
 
+	datasourceSvc, err := apirepository.NewDatasourceService(connectorRepo, catalogProvider)
+	if err != nil {
+		syncService.Stop(ctx)
+
+		return apiserver.APIServerOptions{}, nil, fmt.Errorf("failed to initialize datasource service: %w", err)
+	}
+
 	tokenMgr := auth.NewTokenManager(secretKey, accessTTL, refreshTTL)
 	workerRepo := repository.NewWorkerRepository(pool)
 	workerReg := workerregistry.New(workerRepo)
@@ -117,6 +125,7 @@ func buildAPIServerOptions(ctx context.Context, pool *pgxpool.Pool, secretKey, a
 		TokenManager:       tokenMgr,
 		Blacklist:          blacklist,
 		ApplicationService: apirepository.NewApplicationService(appRepo, svcRepo, compRepo, svcDepRepo, catalogProvider, vars.RuntimeFactory.GetRuntimeType()),
+		DatasourceService:  datasourceSvc,
 		BundleService:      bundlesvc.NewBundleService(bundleRepo, svcRepo, compRepo, catalogProvider),
 		CatalogProvider:    catalogProvider,
 		WorkerGatewayPort:  workerGatewayPort,
diff --git a/ai-services/docs/docs.go b/ai-services/docs/docs.go
index fcfda2fdc..336cd78bc 100644
--- a/ai-services/docs/docs.go
+++ b/ai-services/docs/docs.go
@@ -1366,6 +1366,81 @@ const docTemplate = `{
                 }
             }
         },
+        "/datasources": {
+            "post": {
+                "security": [
+                    {
+                        "BearerAuth": []
+                    }
+                ],
+                "description": "Validates the request, tests the connection, encrypts credentials, and persists a new datasource connector. Returns 422 if the connection test fails.",
+                "consumes": [
+                    "application/json"
+                ],
+                "produces": [
+                    "application/json"
+                ],
+                "tags": [
+                    "Datasources"
+                ],
+                "summary": "Create datasource connector",
+                "parameters": [
+                    {
+                        "description": "Datasource creation request",
+                        "name": "request",
+                        "in": "body",
+                        "required": true,
+                        "schema": {
+                            "$ref": "#/definitions/github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.CreateDatasourceRequest"
+                        }
+                    }
+                ],
+                "responses": {
+                    "201": {
+                        "description": "Datasource created",
+                        "schema": {
+                            "$ref": "#/definitions/github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.CreateDatasourceResponse"
+                        }
+                    },
+                    "400": {
+                        "description": "Invalid request body or validation errors",
+                        "schema": {
+                            "$ref": "#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse"
+                        }
+                    },
+                    "401": {
+                        "description": "Unauthorized",
+                        "schema": {
+                            "$ref": "#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse"
+                        }
+                    },
+                    "404": {
+                        "description": "Provider not found in catalog",
+                        "schema": {
+                            "$ref": "#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse"
+                        }
+                    },
+                    "409": {
+                        "description": "Datasource name already exists",
+                        "schema": {
+                            "$ref": "#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse"
+                        }
+                    },
+                    "422": {
+                        "description": "Connection test failed",
+                        "schema": {
+                            "$ref": "#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse"
+                        }
+                    },
+                    "500": {
+                        "description": "Internal Server Error",
+                        "schema": {
+                            "$ref": "#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse"
+                        }
+                    }
+                }
+            }
+        },
         "/resources": {
             "get": {
                 "security": [
@@ -1805,6 +1880,39 @@ const docTemplate = `{
                 }
             }
         },
+        "github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.CreateDatasourceRequest": {
+            "type": "object",
+            "required": [
+                "name",
+                "params",
+                "provider_id"
+            ],
+            "properties": {
+                "name": {
+                    "description": "Name is the unique human-readable label for this connector.\nMust be 3–100 characters; only letters, digits, hyphens (-), and underscores (_) are\nallowed. Duplicate-name detection is case-insensitive (\"My-DB\" and \"my-db\" conflict).",
+                    "type": "string",
+                    "maxLength": 100,
+                    "minLength": 3
+                },
+                "params": {
+                    "description": "Params holds the provider-specific configuration. Sensitive fields (format: \"password\"\nin the JSON schema) are encrypted at rest; all other fields are stored in plain text.",
+                    "type": "object",
+                    "additionalProperties": {}
+                },
+                "provider_id": {
+                    "description": "ProviderID identifies the provider implementation (e.g. \"object_storage\", \"file_system\").",
+                    "type": "string"
+                }
+            }
+        },
+        "github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.CreateDatasourceResponse": {
+            "type": "object",
+            "properties": {
+                "id": {
+                    "type": "string"
+                }
+            }
+        },
         "github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.Service": {
             "type": "object",
             "required": [
diff --git a/ai-services/docs/swagger.json b/ai-services/docs/swagger.json
index 432b53899..014d636c4 100644
--- a/ai-services/docs/swagger.json
+++ b/ai-services/docs/swagger.json
@@ -1360,6 +1360,81 @@
                 }
             }
         },
+        "/datasources": {
+            "post": {
+                "security": [
+                    {
+                        "BearerAuth": []
+                    }
+                ],
+                "description": "Validates the request, tests the connection, encrypts credentials, and persists a new datasource connector. Returns 422 if the connection test fails.",
+                "consumes": [
+                    "application/json"
+                ],
+                "produces": [
+                    "application/json"
+                ],
+                "tags": [
+                    "Datasources"
+                ],
+                "summary": "Create datasource connector",
+                "parameters": [
+                    {
+                        "description": "Datasource creation request",
+                        "name": "request",
+                        "in": "body",
+                        "required": true,
+                        "schema": {
+                            "$ref": "#/definitions/github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.CreateDatasourceRequest"
+                        }
+                    }
+                ],
+                "responses": {
+                    "201": {
+                        "description": "Datasource created",
+                        "schema": {
+                            "$ref": "#/definitions/github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.CreateDatasourceResponse"
+                        }
+                    },
+                    "400": {
+                        "description": "Invalid request body or validation errors",
+                        "schema": {
+                            "$ref": "#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse"
+                        }
+                    },
+                    "401": {
+                        "description": "Unauthorized",
+                        "schema": {
+                            "$ref": "#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse"
+                        }
+                    },
+                    "404": {
+                        "description": "Provider not found in catalog",
+                        "schema": {
+                            "$ref": "#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse"
+                        }
+                    },
+                    "409": {
+                        "description": "Datasource name already exists",
+                        "schema": {
+                            "$ref": "#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse"
+                        }
+                    },
+                    "422": {
+                        "description": "Connection test failed",
+                        "schema": {
+                            "$ref": "#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse"
+                        }
+                    },
+                    "500": {
+                        "description": "Internal Server Error",
+                        "schema": {
+                            "$ref": "#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse"
+                        }
+                    }
+                }
+            }
+        },
         "/resources": {
             "get": {
                 "security": [
@@ -1799,6 +1874,39 @@
                 }
             }
         },
+        "github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.CreateDatasourceRequest": {
+            "type": "object",
+            "required": [
+                "name",
+                "params",
+                "provider_id"
+            ],
+            "properties": {
+                "name": {
+                    "description": "Name is the unique human-readable label for this connector.\nMust be 3–100 characters; only letters, digits, hyphens (-), and underscores (_) are\nallowed. Duplicate-name detection is case-insensitive (\"My-DB\" and \"my-db\" conflict).",
+                    "type": "string",
+                    "maxLength": 100,
+                    "minLength": 3
+                },
+                "params": {
+                    "description": "Params holds the provider-specific configuration. Sensitive fields (format: \"password\"\nin the JSON schema) are encrypted at rest; all other fields are stored in plain text.",
+                    "type": "object",
+                    "additionalProperties": {}
+                },
+                "provider_id": {
+                    "description": "ProviderID identifies the provider implementation (e.g. \"object_storage\", \"file_system\").",
+                    "type": "string"
+                }
+            }
+        },
+        "github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.CreateDatasourceResponse": {
+            "type": "object",
+            "properties": {
+                "id": {
+                    "type": "string"
+                }
+            }
+        },
         "github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.Service": {
             "type": "object",
             "required": [
diff --git a/ai-services/docs/swagger.yaml b/ai-services/docs/swagger.yaml
index a9a8bf524..1f9af2ecb 100644
--- a/ai-services/docs/swagger.yaml
+++ b/ai-services/docs/swagger.yaml
@@ -41,6 +41,36 @@ definitions:
       id:
         type: string
     type: object
+  github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.CreateDatasourceRequest:
+    properties:
+      name:
+        description: |-
+          Name is the unique human-readable label for this connector.
+          Must be 3–100 characters; only letters, digits, hyphens (-), and underscores (_) are
+          allowed. Duplicate-name detection is case-insensitive ("My-DB" and "my-db" conflict).
+        maxLength: 100
+        minLength: 3
+        type: string
+      params:
+        additionalProperties: {}
+        description: |-
+          Params holds the provider-specific configuration. Sensitive fields (format: "password"
+          in the JSON schema) are encrypted at rest; all other fields are stored in plain text.
+        type: object
+      provider_id:
+        description: ProviderID identifies the provider implementation (e.g. "object_storage",
+          "file_system").
+        type: string
+    required:
+    - name
+    - params
+    - provider_id
+    type: object
+  github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.CreateDatasourceResponse:
+    properties:
+      id:
+        type: string
+    type: object
   github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.Service:
     properties:
       catalog_id:
@@ -1539,6 +1569,56 @@ paths:
       summary: Get connector provider parameters
       tags:
       - Catalog
+  /datasources:
+    post:
+      consumes:
+      - application/json
+      description: Validates the request, tests the connection, encrypts credentials,
+        and persists a new datasource connector. Returns 422 if the connection test
+        fails.
+      parameters:
+      - description: Datasource creation request
+        in: body
+        name: request
+        required: true
+        schema:
+          $ref: '#/definitions/github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.CreateDatasourceRequest'
+      produces:
+      - application/json
+      responses:
+        "201":
+          description: Datasource created
+          schema:
+            $ref: '#/definitions/github_com_project-ai-services_ai-services_internal_pkg_catalog_apiserver_models.CreateDatasourceResponse'
+        "400":
+          description: Invalid request body or validation errors
+          schema:
+            $ref: '#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse'
+        "401":
+          description: Unauthorized
+          schema:
+            $ref: '#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse'
+        "404":
+          description: Provider not found in catalog
+          schema:
+            $ref: '#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse'
+        "409":
+          description: Datasource name already exists
+          schema:
+            $ref: '#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse'
+        "422":
+          description: Connection test failed
+          schema:
+            $ref: '#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse'
+        "500":
+          description: Internal Server Error
+          schema:
+            $ref: '#/definitions/internal_pkg_catalog_apiserver_handlers.ErrorResponse'
+      security:
+      - BearerAuth: []
+      summary: Create datasource connector
+      tags:
+      - Datasources
   /resources:
     get:
       description: Retrieves system resource information including CPU, memory, and
diff --git a/ai-services/go.mod b/ai-services/go.mod
index 18839df23..a56f1b2c6 100644
--- a/ai-services/go.mod
+++ b/ai-services/go.mod
@@ -3,6 +3,10 @@ module github.com/project-ai-services/ai-services
 go 1.26
 
 require (
+	github.com/aws/aws-sdk-go-v2 v1.43.6
+	github.com/aws/aws-sdk-go-v2/credentials v1.19.16
+	github.com/aws/aws-sdk-go-v2/service/s3 v1.107.2
+	github.com/aws/smithy-go v1.27.8
 	github.com/charmbracelet/bubbles v0.21.0
 	github.com/charmbracelet/huh v0.7.0
 	github.com/charmbracelet/lipgloss v1.1.0
@@ -20,6 +24,7 @@ require (
 	github.com/openshift/api v0.0.0-20260213123447-0246c0ac1a77
 	github.com/openshift/client-go v0.0.0-20260213141500-06efc6dce93b
 	github.com/operator-framework/api v0.39.0
+	github.com/pkg/sftp v1.13.9
 	github.com/pressly/goose/v3 v3.27.1
 	github.com/santhosh-tekuri/jsonschema/v6 v6.0.2
 	github.com/spf13/cobra v1.10.2
@@ -61,6 +66,14 @@ require (
 	github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d // indirect
 	github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect
 	github.com/atotto/clipboard v0.1.4 // indirect
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.18 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.37 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.37 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.38 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.17 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.30 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.37 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.38 // indirect
 	github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver/v4 v4.0.0 // indirect
@@ -223,7 +236,6 @@ require (
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
 	github.com/peterbourgon/diskv v2.0.1+incompatible // indirect
 	github.com/pkg/errors v0.9.1 // indirect
-	github.com/pkg/sftp v1.13.9 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/proglottis/gpgme v0.1.5 // indirect
 	github.com/prometheus/client_golang v1.23.2 // indirect
diff --git a/ai-services/go.sum b/ai-services/go.sum
index 9802f0f9a..3f5732f40 100644
--- a/ai-services/go.sum
+++ b/ai-services/go.sum
@@ -41,6 +41,30 @@ github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 h1:DklsrG3d
 github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2/go.mod h1:WaHUgvxTVq04UNunO+XhnAqY/wQc+bxr74GqbsZ/Jqw=
 github.com/atotto/clipboard v0.1.4 h1:EH0zSVneZPSuFR11BlR9YppQTVDbh5+16AmcJi4g1z4=
 github.com/atotto/clipboard v0.1.4/go.mod h1:ZY9tmq7sm5xIbd9bOK4onWV4S6X0u6GY7Vn0Yu86PYI=
+github.com/aws/aws-sdk-go-v2 v1.43.6 h1:RrmFcqCBxkJuf7g1axVo5krB4jM/AO8r5e5oujrgdoQ=
+github.com/aws/aws-sdk-go-v2 v1.43.6/go.mod h1:tXpPM+v0D1lndmga+HqqLDIzUFJlEeR21aspVklHF00=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.18 h1:LAfOuhAH331fmOjTQpAaOlH+Ftn7RzSDJ2VFwjdMMy4=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.18/go.mod h1:4e5xhuXHx1e4U9EthvbPP1r/DIMp5c2823OL8karzcM=
+github.com/aws/aws-sdk-go-v2/credentials v1.19.16 h1:r3RJBuU7X9ibt8RHbMjWE6y60QbKBiII6wSrXnapxSU=
+github.com/aws/aws-sdk-go-v2/credentials v1.19.16/go.mod h1:6cx7zqDENJDbBIIWX6P8s0h6hqHC8Avbjh9Dseo27ug=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.37 h1:lznzIOvvbqjfe8UAaciCRJgBgJsxuTROKlhZuXQWfv8=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.37/go.mod h1:otfkzyfQeMMLZAqX59GSXTL3o22BR/l6HFaRzzbWSqA=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.37 h1:zCEORWo0eU0gDjG+IyApE/2B+ZGG1m+GU7B263XV8ds=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.37/go.mod h1:i6c0PEl3TNOWxRbQ++KQcVenPWS/GoQeiklKhNuqzJ8=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.38 h1:A3UAuCmx7LyUcrixBTzKJYYIUZ2yTvn6ZhT8PB+7APk=
+github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.38/go.mod h1:1PDUYG9Z+JrbbsobsAZHjWOm9QBT/djiK3QbykTL5Z4=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.17 h1:OvYZOB3qA6zvfdRFiRFRzVSiElMYrz3GdntkXZxlp1o=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.17/go.mod h1:JgR/2Ew50ACfIWau1oeMRX59tMtC0kM+PYQGEaT04cY=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.30 h1:5437eMoOwqqQpZn2XJy74mlDCuPYL81texMT3mXqgtU=
+github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.30/go.mod h1:xfu2m3dOpvW8lj98wQYa8V9ku/Rta59hsbireGzhh3A=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.37 h1:a3D4AjrOrTrP8+d9ILBthqrElf0z1JNol09Xvnwcys8=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.37/go.mod h1:ky0gTu+ukvUTuUKFIpp6Wid4oninrkCyvbFkVs0kpHM=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.38 h1:gX8B8y3Ho30B1LPxefDKMi/HZqWEb47U9ogs3DtSG0M=
+github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.38/go.mod h1:l5WblZlcmGPe4/O7JY2HO25Z+xqTBvyfTyFbRMf8gYw=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.107.2 h1:GNU0/xtPEXMKilJZ/a8BedeuQnvu+Usi6qVm9EFfncc=
+github.com/aws/aws-sdk-go-v2/service/s3 v1.107.2/go.mod h1:4jYWUecEsQtE73jPl7p3jrbYXH5ffcR4gegyCygagfg=
+github.com/aws/smithy-go v1.27.8 h1:FR0dxZfIlV7Z8eh2iHfIofdunw382XsDV3Mxt9nUvRY=
+github.com/aws/smithy-go v1.27.8/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc=
 github.com/aymanbagabas/go-osc52/v2 v2.0.1 h1:HwpRHbFMcZLEVr42D4p7XBqjyuxQH5SMiErDT4WkJ2k=
 github.com/aymanbagabas/go-osc52/v2 v2.0.1/go.mod h1:uYgXzlJ7ZpABp8OJ+exZzJJhRNQ2ASbcXHWsFqH8hp8=
 github.com/aymanbagabas/go-udiff v0.2.0 h1:TK0fH4MteXUDspT88n8CKzvK0X9O2xu9yQjWpi6yML8=
diff --git a/ai-services/internal/pkg/catalog/apiserver/apiserver.go b/ai-services/internal/pkg/catalog/apiserver/apiserver.go
index 325044252..beeb78458 100644
--- a/ai-services/internal/pkg/catalog/apiserver/apiserver.go
+++ b/ai-services/internal/pkg/catalog/apiserver/apiserver.go
@@ -52,6 +52,7 @@ type APIServerOptions struct {
 	TokenManager       *auth.TokenManager
 	Blacklist          repository.TokenBlacklist
 	ApplicationService repository.ApplicationServiceInterface
+	DatasourceService  repository.DatasourceServiceInterface
 	BundleService      bundlesvc.BundleServiceInterface
 	CatalogProvider    *catalog.CatalogProvider
 
@@ -70,6 +71,7 @@ type APIserver struct {
 	tokenManager       *auth.TokenManager
 	blacklist          repository.TokenBlacklist
 	applicationService repository.ApplicationServiceInterface
+	datasourceService  repository.DatasourceServiceInterface
 	bundleService      bundlesvc.BundleServiceInterface
 	catalogProvider    *catalog.CatalogProvider
 
@@ -93,6 +95,7 @@ func NewAPIserver(options APIServerOptions) *APIserver {
 		tokenManager:       options.TokenManager,
 		blacklist:          options.Blacklist,
 		applicationService: options.ApplicationService,
+		datasourceService:  options.DatasourceService,
 		bundleService:      options.BundleService,
 		catalogProvider:    options.CatalogProvider,
 		workerGatewayPort:  options.WorkerGatewayPort,
@@ -117,7 +120,7 @@ func (a *APIserver) Start(ctx context.Context) error {
 	}
 	logger.InfofCtx(ctx, "Worker gateway started on %s", gatewayAddr)
 
-	r := CreateRouter(a.authService, a.tokenManager, a.blacklist, a.applicationService, a.workerRegistry, a.bundleService, a.catalogProvider)
+	r := CreateRouter(a.authService, a.tokenManager, a.blacklist, a.applicationService, a.workerRegistry, a.datasourceService, a.bundleService, a.catalogProvider)
 
 	if err := r.Run(fmt.Sprintf(":%d", a.port)); err != nil {
 		return err
diff --git a/ai-services/internal/pkg/catalog/apiserver/handlers/datasource_handler.go b/ai-services/internal/pkg/catalog/apiserver/handlers/datasource_handler.go
new file mode 100644
index 000000000..07077bc03
--- /dev/null
+++ b/ai-services/internal/pkg/catalog/apiserver/handlers/datasource_handler.go
@@ -0,0 +1,83 @@
+package handlers
+
+import (
+	"fmt"
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"github.com/project-ai-services/ai-services/internal/pkg/catalog/apiserver/middleware"
+	"github.com/project-ai-services/ai-services/internal/pkg/catalog/apiserver/models"
+	"github.com/project-ai-services/ai-services/internal/pkg/catalog/apiserver/repository"
+	"github.com/project-ai-services/ai-services/internal/pkg/logger"
+)
+
+// DatasourceHandler handles datasource connector HTTP requests.
+type DatasourceHandler struct {
+	datasourceSvc repository.DatasourceServiceInterface
+}
+
+// NewDatasourceHandler creates a new DatasourceHandler.
+func NewDatasourceHandler(datasourceSvc repository.DatasourceServiceInterface) *DatasourceHandler {
+	return &DatasourceHandler{datasourceSvc: datasourceSvc}
+}
+
+// CreateDatasource godoc
+//
+//	@Summary		Create datasource connector
+//	@Description	Validates the request, tests the connection, encrypts credentials, and persists a new datasource connector. Returns 422 if the connection test fails.
+//	@Tags			Datasources
+//	@Accept			json
+//	@Produce		json
+//	@Security		BearerAuth
+//	@Param			request	body		models.CreateDatasourceRequest	true	"Datasource creation request"
+//	@Success		201		{object}	models.CreateDatasourceResponse	"Datasource created"
+//	@Failure		400		{object}	ErrorResponse					"Invalid request body or validation errors"
+//	@Failure		401		{object}	ErrorResponse					"Unauthorized"
+//	@Failure		404		{object}	ErrorResponse					"Provider not found in catalog"
+//	@Failure		409		{object}	ErrorResponse					"Datasource name already exists"
+//	@Failure		422		{object}	ErrorResponse					"Connection test failed"
+//	@Failure		500		{object}	ErrorResponse					"Internal Server Error"
+//	@Router			/datasources [post]
+func (h *DatasourceHandler) CreateDatasource(c *gin.Context) {
+	var req models.CreateDatasourceRequest
+
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, ErrorResponse{
+			Error: fmt.Sprintf("Invalid request body: %v", err),
+		})
+
+		return
+	}
+
+	// Extract authenticated user from context.
+	userID := c.GetString(middleware.CtxUserIDKey)
+	if userID == "" {
+		c.JSON(http.StatusUnauthorized, ErrorResponse{
+			Error: "Unauthorized: user ID not found in context",
+		})
+
+		return
+	}
+
+	req.CreatedBy = userID
+
+	resp, err := h.datasourceSvc.CreateDatasource(c.Request.Context(), req)
+	if err != nil {
+		if valErr, ok := err.(*repository.ValidationError); ok {
+			c.JSON(valErr.Code, ErrorResponse{Error: valErr.Message})
+
+			return
+		}
+
+		logger.ErrorfCtx(c.Request.Context(), "failed to create datasource: %v", err)
+		c.JSON(http.StatusInternalServerError, ErrorResponse{
+			Error: fmt.Sprintf("Failed to create datasource: %v", err),
+		})
+
+		return
+	}
+
+	c.JSON(http.StatusCreated, resp)
+}
+
+// Made with Bob
diff --git a/ai-services/internal/pkg/catalog/apiserver/models/create_datasource.go b/ai-services/internal/pkg/catalog/apiserver/models/create_datasource.go
new file mode 100644
index 000000000..576af53d8
--- /dev/null
+++ b/ai-services/internal/pkg/catalog/apiserver/models/create_datasource.go
@@ -0,0 +1,23 @@
+package models
+
+// CreateDatasourceRequest is the request body for creating a new datasource connector.
+type CreateDatasourceRequest struct {
+	// Name is the unique human-readable label for this connector.
+	// Must be 3–100 characters; only letters, digits, hyphens (-), and underscores (_) are
+	// allowed. Duplicate-name detection is case-insensitive ("My-DB" and "my-db" conflict).
+	Name string `json:"name" binding:"required,min=3,max=100"`
+	// ProviderID identifies the provider implementation (e.g. "object_storage", "file_system").
+	ProviderID string `json:"provider_id" binding:"required"`
+	// Params holds the provider-specific configuration. Sensitive fields (format: "password"
+	// in the JSON schema) are encrypted at rest; all other fields are stored in plain text.
+	Params map[string]any `json:"params" binding:"required"`
+	// CreatedBy is set from the auth context, never from the request body.
+	CreatedBy string `json:"-"`
+}
+
+// CreateDatasourceResponse is the response body returned after a successful datasource creation.
+type CreateDatasourceResponse struct {
+	ID string `json:"id"`
+}
+
+// Made with Bob
diff --git a/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service.go b/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service.go
new file mode 100644
index 000000000..4734dd71a
--- /dev/null
+++ b/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service.go
@@ -0,0 +1,31 @@
+package repository
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/project-ai-services/ai-services/internal/pkg/catalog"
+	datasourceservice "github.com/project-ai-services/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service"
+	catalogconstants "github.com/project-ai-services/ai-services/internal/pkg/catalog/constants"
+	dbrepo "github.com/project-ai-services/ai-services/internal/pkg/catalog/db/repository"
+	"github.com/project-ai-services/ai-services/internal/pkg/catalog/validators"
+)
+
+// NewDatasourceService creates the DatasourceService wired with all known provider testers.
+// The DB_ENCRYPTION_KEY environment variable is read once at startup and validated immediately
+// so that a missing or empty key causes a fast failure before the server accepts any requests.
+func NewDatasourceService(
+	connectorRepo dbrepo.ConnectorRepository,
+	provider *catalog.CatalogProvider,
+) (DatasourceServiceInterface, error) {
+	encryptionKey := os.Getenv(catalogconstants.DBEncryptionKeyEnv)
+	if encryptionKey == "" {
+		return nil, fmt.Errorf("%s environment variable must be set and non-empty", catalogconstants.DBEncryptionKeyEnv)
+	}
+
+	validator := validators.NewConnectorValidator(provider)
+
+	return datasourceservice.NewDatasourceService(connectorRepo, validator, provider, encryptionKey), nil
+}
+
+// Made with Bob
diff --git a/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service/datasource_service.go b/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service/datasource_service.go
new file mode 100644
index 000000000..40c4eec25
--- /dev/null
+++ b/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service/datasource_service.go
@@ -0,0 +1,169 @@
+package datasourceservice
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/project-ai-services/ai-services/internal/pkg/catalog"
+	apimodels "github.com/project-ai-services/ai-services/internal/pkg/catalog/apiserver/models"
+	catalogconstants "github.com/project-ai-services/ai-services/internal/pkg/catalog/constants"
+	dbmodels "github.com/project-ai-services/ai-services/internal/pkg/catalog/db/models"
+	dbrepo "github.com/project-ai-services/ai-services/internal/pkg/catalog/db/repository"
+	catalogutils "github.com/project-ai-services/ai-services/internal/pkg/catalog/utils"
+	"github.com/project-ai-services/ai-services/internal/pkg/catalog/validators"
+)
+
+const (
+	// ErrMsgDatasourceNameExists is returned when a connector with the given name already exists.
+	ErrMsgDatasourceNameExists = "Datasource with name %q already exists"
+)
+
+// ValidationError re-exported so callers use the same type as for application errors.
+type ValidationError = validators.ValidationError
+
+// DatasourceService is the single implementation of the create-datasource flow.
+// It is provider-agnostic: provider-specific behaviour (connection testing) is
+// delegated to a ConnectionTester looked up from the testers registry.
+// Sensitive-field identification is derived at runtime from each provider's
+// schema.json, keyed on format: "password".
+type DatasourceService struct {
+	connectorRepo   dbrepo.ConnectorRepository
+	validator       *validators.ConnectorValidator
+	catalogProvider *catalog.CatalogProvider
+	encryptionKey   string
+	// testers maps providerID → ConnectionTester. Populated by NewDatasourceService.
+	testers map[string]ConnectionTester
+}
+
+// NewDatasourceService creates a DatasourceService wired with all known provider testers.
+// encryptionKey is the AES-256 key used to encrypt sensitive credential fields; it is
+// injected by the caller (read from DB_ENCRYPTION_KEY at startup) rather than fetched
+// from the environment at call time.
+func NewDatasourceService(
+	connectorRepo dbrepo.ConnectorRepository,
+	validator *validators.ConnectorValidator,
+	catalogProvider *catalog.CatalogProvider,
+	encryptionKey string,
+) *DatasourceService {
+	return &DatasourceService{
+		connectorRepo:   connectorRepo,
+		validator:       validator,
+		catalogProvider: catalogProvider,
+		encryptionKey:   encryptionKey,
+		testers: map[string]ConnectionTester{
+			catalogconstants.DatasourceProviderObjectStorage: NewObjectStorageTester(),
+			catalogconstants.DatasourceProviderFileSystem:    NewFileSystemTester(),
+		},
+	}
+}
+
+// CreateDatasource is the single create flow shared by all providers:
+//
+//  1. Validate the request body (provider existence + JSON-schema param validation).
+//  2. Duplicate-name guard (case-insensitive).
+//  3. Test the connection — the outcome sets the initial connector status.
+//  4. Encrypt sensitive credential fields derived from the provider's schema.json.
+//  5. Persist the connector record.
+func (s *DatasourceService) CreateDatasource(ctx context.Context, req apimodels.CreateDatasourceRequest) (*apimodels.CreateDatasourceResponse, error) {
+	// Phase 1: validate request (provider existence + param schema).
+	if err := s.validator.ValidateCreateDatasourceRequest(ctx, req); err != nil {
+		return nil, err
+	}
+
+	// Phase 2: duplicate-name guard (case-insensitive — handled by LOWER() in the DB query).
+	existing, err := s.connectorRepo.GetByName(ctx, req.Name)
+	if err != nil {
+		return nil, fmt.Errorf("failed to check for existing connector: %w", err)
+	}
+	if existing != nil {
+		return nil, &ValidationError{
+			Code:    http.StatusConflict,
+			Message: fmt.Sprintf(ErrMsgDatasourceNameExists, req.Name),
+		}
+	}
+
+	// Phase 3: test connection — determines the initial connector status.
+	tester, ok := s.testers[req.ProviderID]
+	if !ok {
+		// Should not happen after Phase 1 validation, but guard defensively.
+		return nil, &ValidationError{
+			Code:    http.StatusBadRequest,
+			Message: fmt.Sprintf("No connection tester registered for provider %q", req.ProviderID),
+		}
+	}
+
+	testErr := tester.TestConnection(ctx, req.Params)
+	if testErr != nil {
+		return nil, &ValidationError{
+			Code:    http.StatusUnprocessableEntity,
+			Message: fmt.Sprintf("Connection test failed: %v", testErr),
+		}
+	}
+
+	// Phase 4: derive sensitive fields from the provider's schema.json and encrypt.
+	schema, err := s.catalogProvider.GetConnectorProviderParams(ctx, catalogconstants.ConnectorTypeDatasource, req.ProviderID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load schema for provider %q: %w", req.ProviderID, err)
+	}
+
+	encryptedParams, err := encryptSensitiveFields(req.Params, sensitiveFieldsFromSchema(schema), s.encryptionKey)
+	if err != nil {
+		return nil, fmt.Errorf("failed to encrypt connector credentials: %w", err)
+	}
+
+	// Phase 5: persist the connector record.
+	connector := &dbmodels.Connector{
+		Name:      req.Name,
+		Type:      catalogconstants.ConnectorTypeDatasource,
+		Provider:  req.ProviderID,
+		Status:    dbmodels.ConnectorStatusConnected,
+		Metadata:  encryptedParams,
+		CreatedBy: req.CreatedBy,
+	}
+
+	if err := s.connectorRepo.Insert(ctx, connector); err != nil {
+		return nil, fmt.Errorf("failed to persist connector: %w", err)
+	}
+
+	return &apimodels.CreateDatasourceResponse{ID: connector.ID.String()}, nil
+}
+
+// encryptSensitiveFields returns a copy of params where every key listed in
+// sensitiveKeys has its string value replaced with an AES-256-GCM ciphertext.
+// encryptionKey is the AES-256 secret injected at service construction time (DB_ENCRYPTION_KEY).
+func encryptSensitiveFields(params map[string]any, sensitiveKeys map[string]bool, encryptionKey string) (map[string]any, error) {
+	if len(sensitiveKeys) == 0 {
+		return params, nil
+	}
+
+	if encryptionKey == "" {
+		return nil, fmt.Errorf("encryption key is not configured (DB_ENCRYPTION_KEY must be set)")
+	}
+
+	result := make(map[string]any, len(params))
+	for k, v := range params {
+		if sensitiveKeys[k] {
+			plaintext, ok := v.(string)
+			if !ok {
+				return nil, &ValidationError{
+					Code:    http.StatusBadRequest,
+					Message: fmt.Sprintf("sensitive field %q must be a string value", k),
+				}
+			}
+
+			ciphertext, err := catalogutils.Encrypt(plaintext, encryptionKey)
+			if err != nil {
+				return nil, fmt.Errorf("failed to encrypt field %q: %w", k, err)
+			}
+
+			result[k] = ciphertext
+		} else {
+			result[k] = v
+		}
+	}
+
+	return result, nil
+}
+
+// Made with Bob
diff --git a/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service/file_system.go b/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service/file_system.go
new file mode 100644
index 000000000..2312c5e18
--- /dev/null
+++ b/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service/file_system.go
@@ -0,0 +1,132 @@
+package datasourceservice
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/pkg/sftp"
+	"golang.org/x/crypto/ssh"
+)
+
+const (
+	sshDialTimeout = 10 * time.Second
+)
+
+// fileSystemTester implements ConnectionTester for SSH/SFTP-based file system sources.
+type fileSystemTester struct{}
+
+// NewFileSystemTester returns a ConnectionTester for SSH/SFTP file system providers.
+func NewFileSystemTester() ConnectionTester {
+	return &fileSystemTester{}
+}
+
+// TestConnection runs three sequential checks against an SSH/SFTP endpoint:
+//  1. Network — TCP dial to host (port defaults to 22 when not supplied).
+//  2. Auth    — SSH handshake using the PEM private key.
+//  3. Access  — SFTP Stat on remote_path.
+//
+// The overall context deadline is propagated to the raw TCP connection so that
+// every subsequent operation (SSH handshake, SFTP subsystem, Stat) is
+// automatically cancelled when the deadline expires.
+func (t *fileSystemTester) TestConnection(ctx context.Context, params map[string]any) error {
+	host, _ := params["host"].(string)
+	username, _ := params["username"].(string)
+	privateKeyPEM, _ := params["private_key"].(string)
+	remotePath, _ := params["remote_path"].(string)
+
+	// Port is not a required schema field; default to 22 when absent.
+	port := "22"
+	if p, ok := params["port"].(string); ok && p != "" {
+		port = p
+	}
+
+	// ── 1. Network ────────────────────────────────────────────────────────────
+	addr := net.JoinHostPort(host, port)
+	dialer := &net.Dialer{Timeout: sshDialTimeout}
+
+	tcpConn, err := dialer.DialContext(ctx, "tcp", addr)
+	if err != nil {
+		return &ConnectionCheckError{
+			CheckType: ConnectionCheckNetwork,
+			Message:   fmt.Sprintf("TCP dial %s failed: %v", addr, err),
+		}
+	}
+
+	// Apply the remaining context deadline to the raw connection so every
+	// subsequent operation (SSH handshake, SFTP I/O) is bounded by it.
+	if deadline, ok := ctx.Deadline(); ok {
+		_ = tcpConn.SetDeadline(deadline)
+	}
+
+	// ── Parse private key ─────────────────────────────────────────────────────
+	signer, err := ssh.ParsePrivateKey([]byte(privateKeyPEM))
+	if err != nil {
+		_ = tcpConn.Close()
+
+		return &ConnectionCheckError{
+			CheckType: ConnectionCheckAuth,
+			Message:   "could not parse private_key: ensure it is a valid PEM-encoded key with newlines preserved (\\n between lines)",
+		}
+	}
+
+	// ── 2. Auth — SSH handshake ───────────────────────────────────────────────
+	sshCfg := &ssh.ClientConfig{
+		User:            username,
+		Auth:            []ssh.AuthMethod{ssh.PublicKeys(signer)},
+		HostKeyCallback: ssh.InsecureIgnoreHostKey(), // #nosec G106 — diagnostic probe; not used on the data path
+		Timeout:         sshDialTimeout,
+	}
+
+	sshConn, chans, reqs, err := ssh.NewClientConn(tcpConn, addr, sshCfg)
+	if err != nil {
+		return &ConnectionCheckError{
+			CheckType: ConnectionCheckAuth,
+			Message:   "SSH authentication failed: verify the username and private key are correct for the target host",
+		}
+	}
+
+	sshClient := ssh.NewClient(sshConn, chans, reqs)
+	defer func() { _ = sshClient.Close() }()
+
+	// ── 3. Access — SFTP stat ────────────────────────────────────────────────
+	return checkSFTPAccess(sshClient, remotePath)
+}
+
+// checkSFTPAccess opens an SFTP subsystem and stats the remote path to confirm it
+// exists and is reachable. Stat (SSH_FXP_STAT) is sufficient: a successful response
+// proves the authenticated user has at least execute permission on the path and that
+// the path is accessible — which is the meaningful check for a datasource connection
+// test. ReadDir is intentionally avoided here: it buffers every directory entry from
+// the server before returning, which is unbounded in time for large directories.
+// The underlying TCP connection already has a deadline so all SFTP calls are bounded.
+func checkSFTPAccess(sshClient *ssh.Client, remotePath string) error {
+	sftpClient, err := sftp.NewClient(sshClient)
+	if err != nil {
+		return &ConnectionCheckError{
+			CheckType: ConnectionCheckAccess,
+			Message:   fmt.Sprintf("could not open SFTP subsystem: %v", err),
+		}
+	}
+	defer func() { _ = sftpClient.Close() }()
+
+	info, err := sftpClient.Stat(remotePath)
+	if err != nil {
+		return &ConnectionCheckError{
+			CheckType: ConnectionCheckAccess,
+			Message:   fmt.Sprintf("sftp.Stat(%q) failed: %v", remotePath, err),
+		}
+	}
+
+	if !info.IsDir() {
+		return &ConnectionCheckError{
+			CheckType: ConnectionCheckAccess,
+			Message:   fmt.Sprintf("remote path %q is not a directory", remotePath),
+		}
+	}
+
+	return nil
+}
+
+// Made with Bob
diff --git a/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service/interface.go b/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service/interface.go
new file mode 100644
index 000000000..c864497a4
--- /dev/null
+++ b/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service/interface.go
@@ -0,0 +1,70 @@
+package datasourceservice
+
+import (
+	"context"
+	"fmt"
+	"strings"
+)
+
+// ConnectionCheckType identifies which phase of the connection test failed.
+type ConnectionCheckType string
+
+const (
+	// ConnectionCheckNetwork indicates a TCP/DNS reachability failure.
+	ConnectionCheckNetwork ConnectionCheckType = "network"
+	// ConnectionCheckAuth indicates a credential rejection (invalid key, bad password, etc.).
+	ConnectionCheckAuth ConnectionCheckType = "auth"
+	// ConnectionCheckAccess indicates valid credentials but insufficient permissions or a missing resource.
+	ConnectionCheckAccess ConnectionCheckType = "access"
+)
+
+// ConnectionCheckError is returned when one of the three sequential connection checks fails.
+// It carries the check type and a human-readable message so callers can surface a specific,
+// actionable error to the user.
+type ConnectionCheckError struct {
+	// CheckType is the phase of the test that failed (network, auth, or access).
+	CheckType ConnectionCheckType
+	// Message describes the failure in human-readable terms.
+	Message string
+}
+
+func (e *ConnectionCheckError) Error() string {
+	return fmt.Sprintf("[%s] %s", strings.ToUpper(string(e.CheckType)), e.Message)
+}
+
+// ConnectionTester is the interface implemented by each datasource provider.
+// It captures only the behaviour that differs between providers; the generic
+// CreateDatasource flow (validation, encryption, DB insert) lives in DatasourceService
+// and delegates to this interface for the parts that vary.
+type ConnectionTester interface {
+	// TestConnection runs provider-specific connectivity checks (network → auth → access).
+	// Returns nil when all checks pass, or a *ConnectionCheckError on the first failure.
+	TestConnection(ctx context.Context, params map[string]any) error
+}
+
+// sensitiveFieldsFromSchema inspects the top-level properties of a JSON Schema map and
+// returns the set of property names whose "format" is "password". This allows the set of
+// fields that require encryption to be driven by the connector's schema.json rather than
+// being hardcoded in each provider implementation.
+func sensitiveFieldsFromSchema(schema map[string]any) map[string]bool {
+	sensitive := make(map[string]bool)
+
+	properties, ok := schema["properties"].(map[string]any)
+	if !ok {
+		return sensitive
+	}
+
+	for name, raw := range properties {
+		prop, ok := raw.(map[string]any)
+		if !ok {
+			continue
+		}
+		if fmt, ok := prop["format"].(string); ok && fmt == "password" {
+			sensitive[name] = true
+		}
+	}
+
+	return sensitive
+}
+
+// Made with Bob
diff --git a/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service/object_storage.go b/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service/object_storage.go
new file mode 100644
index 000000000..74e97798c
--- /dev/null
+++ b/ai-services/internal/pkg/catalog/apiserver/repository/datasource_service/object_storage.go
@@ -0,0 +1,192 @@
+package datasourceservice
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"regexp"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awshttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/smithy-go"
+)
+
+const (
+	awsEndpointSuffix = "amazonaws.com"
+	s3ConnectTimeout  = 10 * time.Second
+	s3DefaultRegion   = "us-east-1"
+)
+
+// regionFromEndpointRe extracts the SigV4 region segment from AWS S3 and IBM COS endpoint URLs.
+// It matches the same two hostname patterns as the digitize Python service (_REGION_FROM_URL_RE).
+//
+//	s3.<region>.amazonaws.com                              (AWS S3)
+//	s3.<region>.cloud-object-storage.appdomain.cloud      (IBM COS)
+var regionFromEndpointRe = regexp.MustCompile(
+	`(?i)s3[.\-](?P<region>[a-z0-9\-]+)\.(?:amazonaws\.com|cloud-object-storage\.appdomain\.cloud)`,
+)
+
+// cosCrossRegionAliases maps IBM COS cross-region endpoint aliases to the canonical SigV4
+// region that IBM COS accepts for signing. These three aliases are fixed IBM infrastructure
+// geography mappings (documented by IBM COS) — identical to the Python digitize service.
+//
+//	us → us-south  (Dallas  — primary US cross-region PoP)
+//	eu → eu-de     (Frankfurt — primary EU cross-region PoP)
+//	ap → jp-tok    (Tokyo   — primary AP cross-region PoP)
+var cosCrossRegionAliases = map[string]string{
+	"us": "us-south",
+	"eu": "eu-de",
+	"ap": "jp-tok",
+}
+
+// authErrorCodes are S3/STS codes that indicate invalid credentials.
+var authErrorCodes = map[string]bool{
+	"InvalidAccessKeyId":    true,
+	"SignatureDoesNotMatch": true,
+	"InvalidClientTokenId":  true,
+	"AuthFailure":           true,
+}
+
+// accessErrorCodes are S3 codes that indicate valid credentials but insufficient
+// permissions or a missing/inaccessible bucket.
+var accessErrorCodes = map[string]bool{
+	"AccessDenied":      true,
+	"NoSuchBucket":      true,
+	"AllAccessDisabled": true,
+}
+
+// objectStorageTester implements ConnectionTester for S3-compatible object storage.
+// It handles both AWS S3 and IBM COS (and any other S3-compatible endpoint).
+//
+// AWS S3:  the SDK resolves the regional endpoint from cfg.Region automatically
+//
+//	using virtual-hosted-style addressing; no BaseEndpoint is set.
+//
+// IBM COS / S3-compatible: BaseEndpoint is set to the supplied URL and path-style
+//
+//	addressing is enabled, which is required by most non-AWS stores.
+type objectStorageTester struct{}
+
+// NewObjectStorageTester returns a ConnectionTester for S3-compatible object storage providers.
+func NewObjectStorageTester() ConnectionTester {
+	return &objectStorageTester{}
+}
+
+// TestConnection runs three sequential checks against an S3-compatible endpoint using a
+// single ListObjectsV2(MaxKeys=0) call — a zero-cost probe that transfers no object data.
+// The SDK response (or error) is classified into network / auth / access failure categories.
+func (t *objectStorageTester) TestConnection(ctx context.Context, params map[string]any) error {
+	endpointURL, _ := params["endpoint_url"].(string)
+	bucket, _ := params["bucket_name"].(string)
+	accessKeyID, _ := params["access_key_id"].(string)
+	secretKey, _ := params["secret_access_key"].(string)
+	prefix, _ := params["prefix"].(string)
+
+	cfg := aws.Config{
+		Region:      regionFromEndpoint(endpointURL),
+		Credentials: credentials.NewStaticCredentialsProvider(accessKeyID, secretKey, ""),
+		HTTPClient:  awshttp.NewBuildableClient().WithTimeout(s3ConnectTimeout),
+	}
+
+	client := s3.NewFromConfig(cfg, func(o *s3.Options) {
+		if strings.Contains(endpointURL, awsEndpointSuffix) {
+			// AWS S3: SDK resolves the regional endpoint from cfg.Region automatically;
+			// virtual-hosted-style addressing is used by default — no BaseEndpoint needed.
+			return
+		}
+		// IBM COS, MinIO, and other S3-compatible stores: supply the custom
+		// endpoint and enable path-style addressing.
+		o.BaseEndpoint = aws.String(endpointURL)
+		o.UsePathStyle = true
+	})
+
+	callCtx, cancel := context.WithTimeout(ctx, s3ConnectTimeout)
+	defer cancel()
+
+	maxKeys := int32(0)
+	input := &s3.ListObjectsV2Input{
+		Bucket:  aws.String(bucket),
+		MaxKeys: &maxKeys,
+	}
+	if prefix != "" {
+		input.Prefix = aws.String(prefix)
+	}
+
+	_, err := client.ListObjectsV2(callCtx, input)
+	if err == nil {
+		return nil
+	}
+
+	return classifyS3Error(err, endpointURL, bucket)
+}
+
+// classifyS3Error maps an AWS SDK error to a *ConnectionCheckError for the appropriate phase.
+func classifyS3Error(err error, endpointURL, bucket string) error {
+	// Network error: DNS failure, TCP refused, or context deadline from an unreachable host.
+	var netErr *net.OpError
+	if errors.As(err, &netErr) ||
+		errors.Is(err, context.DeadlineExceeded) ||
+		errors.Is(err, context.Canceled) {
+		return &ConnectionCheckError{
+			CheckType: ConnectionCheckNetwork,
+			Message:   fmt.Sprintf("cannot reach S3 endpoint %q: %v", endpointURL, err),
+		}
+	}
+
+	// Network passed — a response was received; inspect the API error code.
+	var apiErr smithy.APIError
+	if errors.As(err, &apiErr) {
+		code := apiErr.ErrorCode()
+		if authErrorCodes[code] {
+			return &ConnectionCheckError{
+				CheckType: ConnectionCheckAuth,
+				Message:   fmt.Sprintf("credential rejected (code: %s) — check access_key_id / secret_access_key", code),
+			}
+		}
+		if accessErrorCodes[code] {
+			return &ConnectionCheckError{
+				CheckType: ConnectionCheckAccess,
+				Message:   fmt.Sprintf("bucket %q not accessible (code: %s)", bucket, code),
+			}
+		}
+		// Unknown API error — surface code for diagnostics.
+		return &ConnectionCheckError{
+			CheckType: ConnectionCheckAuth,
+			Message:   fmt.Sprintf("unexpected S3 error (code: %s): %v", code, err),
+		}
+	}
+
+	// Non-API error with a network-level response (e.g. TLS failure).
+	return &ConnectionCheckError{
+		CheckType: ConnectionCheckNetwork,
+		Message:   fmt.Sprintf("S3 connectivity error for %q: %v", endpointURL, err),
+	}
+}
+
+// regionFromEndpoint extracts the SigV4 region from an S3-compatible endpoint URL using
+// the same regex-based approach as the digitize Python service (_REGION_FROM_URL_RE in config.py).
+//
+// It matches AWS S3 and IBM COS hostname patterns and resolves IBM COS cross-region aliases
+// to their canonical SigV4 region values. Falls back to s3DefaultRegion ("us-east-1") for
+// any URL that does not match a known pattern (e.g. MinIO, custom S3-compatible stores) —
+// this is the AWS SDK default and is correct for SigV4 signing against most S3-compatible stores.
+func regionFromEndpoint(endpointURL string) string {
+	match := regionFromEndpointRe.FindStringSubmatch(endpointURL)
+	if match == nil {
+		return s3DefaultRegion
+	}
+
+	region := strings.ToLower(match[regionFromEndpointRe.SubexpIndex("region")])
+	if canonical, ok := cosCrossRegionAliases[region]; ok {
+		return canonical
+	}
+
+	return region
+}
+
+// Made with Bob
diff --git a/ai-services/internal/pkg/catalog/apiserver/repository/interface.go b/ai-services/internal/pkg/catalog/apiserver/repository/interface.go
index 89b208a4a..b36e40b84 100644
--- a/ai-services/internal/pkg/catalog/apiserver/repository/interface.go
+++ b/ai-services/internal/pkg/catalog/apiserver/repository/interface.go
@@ -8,6 +8,13 @@ import (
 	"github.com/project-ai-services/ai-services/internal/pkg/catalog/types"
 )
 
+// DatasourceServiceInterface defines the contract for datasource connector business logic.
+type DatasourceServiceInterface interface {
+	// CreateDatasource validates the request, tests the connection, encrypts credentials,
+	// and persists a new datasource connector record.
+	CreateDatasource(ctx context.Context, req apimodels.CreateDatasourceRequest) (*apimodels.CreateDatasourceResponse, error)
+}
+
 // ApplicationServiceInterface defines the contract for application business logic.
 type ApplicationServiceInterface interface {
 	// ListApplications retrieves a paginated list of applications with filters.
diff --git a/ai-services/internal/pkg/catalog/apiserver/router.go b/ai-services/internal/pkg/catalog/apiserver/router.go
index 575f4eb1e..195294190 100644
--- a/ai-services/internal/pkg/catalog/apiserver/router.go
+++ b/ai-services/internal/pkg/catalog/apiserver/router.go
@@ -18,7 +18,7 @@ import (
 )
 
 // CreateRouter sets up the Gin router with the necessary routes and authentication middleware for the API server.
-func CreateRouter(authSvc auth.Service, tokenMgr *auth.TokenManager, blacklist repository.TokenBlacklist, appService repository.ApplicationServiceInterface, workerReg *registry.Registry, bundleService bundlesvc.BundleServiceInterface, catalogProvider *catalog.CatalogProvider) *gin.Engine {
+func CreateRouter(authSvc auth.Service, tokenMgr *auth.TokenManager, blacklist repository.TokenBlacklist, appService repository.ApplicationServiceInterface, workerReg *registry.Registry, datasourceSvc repository.DatasourceServiceInterface, bundleService bundlesvc.BundleServiceInterface, catalogProvider *catalog.CatalogProvider) *gin.Engine {
 	if mode := os.Getenv("GIN_MODE"); mode != "" {
 		gin.SetMode(mode)
 	}
@@ -39,6 +39,7 @@ func CreateRouter(authSvc auth.Service, tokenMgr *auth.TokenManager, blacklist r
 	registerCatalogRoutes(v1, handlers.NewCatalogHandler(catalogProvider), handlers.NewResourcesHandler(), auth)
 	registerApplicationRoutes(v1, handlers.NewApplicationHandler(appService), auth)
 	registerWorkerRoutes(v1, handlers.NewWorkerHandler(workerReg), auth)
+	registerDatasourceRoutes(v1, handlers.NewDatasourceHandler(datasourceSvc), auth)
 	registerBundleRoutes(v1, handlers.NewBundleHandler(bundleService), auth)
 
 	return router
@@ -113,3 +114,11 @@ func registerWorkerRoutes(v1 *gin.RouterGroup, h *handlers.WorkerHandler, authMw
 		g.DELETE("/:id", h.DeleteWorker)
 	}
 }
+
+func registerDatasourceRoutes(v1 *gin.RouterGroup, h *handlers.DatasourceHandler, authMw gin.HandlerFunc) {
+	g := v1.Group("datasources")
+	g.Use(authMw)
+	{
+		g.POST("", h.CreateDatasource)
+	}
+}
diff --git a/ai-services/internal/pkg/catalog/constants/catalog.go b/ai-services/internal/pkg/catalog/constants/catalog.go
index 4d0080296..ddd9a8d96 100644
--- a/ai-services/internal/pkg/catalog/constants/catalog.go
+++ b/ai-services/internal/pkg/catalog/constants/catalog.go
@@ -20,6 +20,22 @@ const (
 	CatalogTypeConnectors = "connectors"
 )
 
+// Connector type constants.
+const (
+	// ConnectorTypeDatasource is the catalog connector type shared by all datasource providers.
+	ConnectorTypeDatasource = "datasource"
+)
+
+// Datasource provider ID constants.
+// These values must match the connector IDs defined in the catalog assets
+// (assets/connectors/datasource/<id>/metadata.yaml).
+const (
+	// DatasourceProviderObjectStorage is the provider ID for S3-compatible object storage connectors.
+	DatasourceProviderObjectStorage = "object_storage"
+	// DatasourceProviderFileSystem is the provider ID for SSH/SFTP file system connectors.
+	DatasourceProviderFileSystem = "file_system"
+)
+
 // Catalog name constants.
 const (
 	// CatalogAppName represents the catalog name.
@@ -70,4 +86,11 @@ const (
 	DefaultHTTPSPort = "443"
 )
 
+// Environment variable name constants.
+const (
+	// DBEncryptionKeyEnv is the environment variable that holds the AES-256 key used to
+	// encrypt sensitive connector credential fields at rest.
+	DBEncryptionKeyEnv = "DB_ENCRYPTION_KEY"
+)
+
 // Made with Bob
diff --git a/ai-services/internal/pkg/catalog/db/repository/connector_repo.go b/ai-services/internal/pkg/catalog/db/repository/connector_repo.go
index a32ccb8e8..65126b8eb 100644
--- a/ai-services/internal/pkg/catalog/db/repository/connector_repo.go
+++ b/ai-services/internal/pkg/catalog/db/repository/connector_repo.go
@@ -35,6 +35,9 @@ type ConnectorUpdateFields struct {
 type ConnectorRepository interface {
 	// Insert creates a new connector, populating the ID, CreatedAt, and UpdatedAt fields on success.
 	Insert(ctx context.Context, connector *models.Connector) error
+	// GetByName retrieves a connector by its unique name.
+	// Returns nil (not an error) when no connector with that name exists.
+	GetByName(ctx context.Context, name string) (*models.Connector, error)
 	// GetByID retrieves a connector by its UUID.
 	// When includeCreds is false the metadata column is omitted (safe for API responses).
 	// When includeCreds is true the full row including metadata is returned; callers are
@@ -160,6 +163,29 @@ func (r *connectorRepo) Insert(ctx context.Context, connector *models.Connector)
 	return nil
 }
 
+// GetByName retrieves a connector by its unique name.
+// The comparison is case-insensitive so that "My-DB" and "my-db" are treated as the same name.
+// Returns nil (not an error) when no connector with that name exists.
+func (r *connectorRepo) GetByName(ctx context.Context, name string) (*models.Connector, error) {
+	query := `SELECT ` + nonSensitiveColumns + ` FROM connectors WHERE LOWER(name) = LOWER($1)`
+
+	rows, err := r.pool.Query(ctx, query, name)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get connector by name: %w", err)
+	}
+	defer rows.Close()
+
+	if !rows.Next() {
+		if err := rows.Err(); err != nil {
+			return nil, fmt.Errorf("failed to get connector by name: %w", err)
+		}
+
+		return nil, nil
+	}
+
+	return scanConnector(rows)
+}
+
 // GetByID retrieves a connector by UUID.
 // Pass includeCreds=false for API responses (metadata omitted).
 // Pass includeCreds=true for internal paths that need credentials (sync job, Digitize propagation);
diff --git a/ai-services/internal/pkg/catalog/validators/validation.go b/ai-services/internal/pkg/catalog/validators/validation.go
index e8c9865a5..f3a02a2c9 100644
--- a/ai-services/internal/pkg/catalog/validators/validation.go
+++ b/ai-services/internal/pkg/catalog/validators/validation.go
@@ -4,11 +4,13 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"regexp"
 	"sort"
 	"strings"
 
 	"github.com/project-ai-services/ai-services/internal/pkg/catalog"
 	apimodels "github.com/project-ai-services/ai-services/internal/pkg/catalog/apiserver/models"
+	catalogconstants "github.com/project-ai-services/ai-services/internal/pkg/catalog/constants"
 	"github.com/project-ai-services/ai-services/internal/pkg/catalog/types"
 	"github.com/project-ai-services/ai-services/internal/pkg/catalog/utils"
 )
@@ -434,4 +436,59 @@ func formatParamKeys(keys []string) string {
 	return strings.Join(quoted, ", ")
 }
 
+// ConnectorValidator validates CreateDatasourceRequest payloads against the catalog.
+type ConnectorValidator struct {
+	provider *catalog.CatalogProvider
+}
+
+// datasourceNameRe restricts connector names to letters, digits, hyphens, and underscores.
+// This matches the character set allowed by most cloud resource naming conventions.
+var datasourceNameRe = regexp.MustCompile(`^[a-zA-Z0-9_-]+$`)
+
+// NewConnectorValidator creates a new ConnectorValidator backed by the given catalog provider.
+func NewConnectorValidator(provider *catalog.CatalogProvider) *ConnectorValidator {
+	return &ConnectorValidator{provider: provider}
+}
+
+// ValidateCreateDatasourceRequest validates the full CreateDatasourceRequest:
+//  1. Validates the name contains only allowed characters (letters, digits, hyphens, underscores).
+//  2. Verifies the provider exists in the catalog under the "datasource" connector type.
+//  3. Validates params against the provider's JSON Schema (if one is present).
+func (v *ConnectorValidator) ValidateCreateDatasourceRequest(ctx context.Context, req apimodels.CreateDatasourceRequest) error {
+	// Name character validation — case-insensitive duplicate detection is handled at
+	// the DB query level via LOWER(name) = LOWER($1).
+	if !datasourceNameRe.MatchString(req.Name) {
+		return &ValidationError{
+			Code:    http.StatusBadRequest,
+			Message: "Datasource name may only contain letters, digits, hyphens (-), and underscores (_)",
+		}
+	}
+
+	if !v.provider.ConnectorExists(catalogconstants.ConnectorTypeDatasource, req.ProviderID) {
+		return &ValidationError{
+			Code:    http.StatusNotFound,
+			Message: fmt.Sprintf("Datasource provider %q not found in catalog", req.ProviderID),
+		}
+	}
+
+	schema, err := v.provider.GetConnectorProviderParams(ctx, catalogconstants.ConnectorTypeDatasource, req.ProviderID)
+	if err != nil {
+		return fmt.Errorf("failed to load param schema for provider %q: %w", req.ProviderID, err)
+	}
+
+	if len(schema) == 0 {
+		// No schema defined for this provider — no parameter constraints to enforce.
+		return nil
+	}
+
+	if len(req.Params) == 0 {
+		return &ValidationError{
+			Code:    http.StatusBadRequest,
+			Message: fmt.Sprintf("Params is required for datasource provider %q", req.ProviderID),
+		}
+	}
+
+	return ValidateParams(req.Params, schema, fmt.Sprintf("datasource provider %q", req.ProviderID))
+}
+
 // Made with Bob