diff --git a/skyvern/exceptions.py b/skyvern/exceptions.py
index d7777474..3818bfab 100644
--- a/skyvern/exceptions.py
+++ b/skyvern/exceptions.py
@@ -163,6 +163,14 @@ class MissingValueForParameter(SkyvernHTTPException):
         )
 
 
+class InvalidCredentialId(SkyvernHTTPException):
+    def __init__(self, credential_id: str) -> None:
+        super().__init__(
+            f"Invalid credential ID: {credential_id}. Failed to resolve to a valid credential.",
+            status_code=status.HTTP_400_BAD_REQUEST,
+        )
+
+
 class WorkflowParameterNotFound(SkyvernHTTPException):
     def __init__(self, workflow_parameter_id: str) -> None:
         super().__init__(
diff --git a/skyvern/forge/sdk/workflow/service.py b/skyvern/forge/sdk/workflow/service.py
index 45819c27..5509a8c2 100644
--- a/skyvern/forge/sdk/workflow/service.py
+++ b/skyvern/forge/sdk/workflow/service.py
@@ -11,6 +11,7 @@ from skyvern.config import settings
 from skyvern.constants import GET_DOWNLOADED_FILES_TIMEOUT, SAVE_DOWNLOADED_FILES_TIMEOUT
 from skyvern.exceptions import (
     FailedToSendWebhook,
+    InvalidCredentialId,
     MissingValueForParameter,
     SkyvernException,
     WorkflowNotFound,
@@ -118,6 +119,11 @@ class WorkflowService:
                 results.extend(WorkflowService._collect_extracted_information(item))
         return results
 
+    async def _validate_credential_id(self, credential_id: str, organization: Organization) -> None:
+        credential = await app.DATABASE.get_credential(credential_id, organization_id=organization.organization_id)
+        if credential is None:
+            raise InvalidCredentialId(credential_id)
+
     async def setup_workflow_run(
         self,
         request_id: str | None,
@@ -189,12 +195,16 @@ class WorkflowService:
             for workflow_parameter in all_workflow_parameters:
                 if workflow_request.data and workflow_parameter.key in workflow_request.data:
                     request_body_value = workflow_request.data[workflow_parameter.key]
+                    if workflow_parameter.workflow_parameter_type == WorkflowParameterType.CREDENTIAL_ID:
+                        await self._validate_credential_id(str(request_body_value), organization)
                     await self.create_workflow_run_parameter(
                         workflow_run_id=workflow_run.workflow_run_id,
                         workflow_parameter=workflow_parameter,
                         value=request_body_value,
                     )
                 elif workflow_parameter.default_value is not None:
+                    if workflow_parameter.workflow_parameter_type == WorkflowParameterType.CREDENTIAL_ID:
+                        await self._validate_credential_id(str(workflow_parameter.default_value), organization)
                     await self.create_workflow_run_parameter(
                         workflow_run_id=workflow_run.workflow_run_id,
                         workflow_parameter=workflow_parameter,