Retrieve credentials
Use the SecretStore service to retrieve credentials from Atlan's secure credential storage during workflow execution. The service resolves credential GUIDs into actual credential values needed for connecting to external systems.
get_credentials()
Asynchronously retrieves credentials from the Secret Store using a credential GUID reference. This method resolves credential GUIDs into actual credential values needed for external connections.
Every workflow begins with get_workflow_config
which retrieves configuration containing the credential GUID reference—never actual credentials. During workflow execution, you use this method to resolve credentials only when needed.
Syntax
await SecretStore.get_credentials(credential_guid)
Parameters
The method accepts a single parameter:
credential_guid
Required
The unique identifier for the credentials stored in the Secret Store. This GUID is automatically provided in workflow_args when the workflow is triggered by the Atlan Orchestrator.
Return value
The method returns a dictionary containing the credential fields required for authentication. The structure varies based on the credential type configured in the Atlan UI.
Type: Dict[str, Any]
{
"username": "db_user",
"password": "secure_password",
"host": "db.example.com",
"port": 5432
}
Usage
This example demonstrates how to retrieve credentials within a workflow activity. The pattern ensures your application code never handles raw credentials directly - the credentials exist only in memory during the specific operation that requires them, and the framework handles all the complexity of secure retrieval and cleanup.
from application_sdk.services.secretstore import SecretStore
from typing import Dict, Any, Optional
import logging
logger = logging.getLogger(__name__)
async def demo_activity(
self, workflow_args: Dict[str, Any]
) -> Optional[ActivityStatistics]:
"""
Example activity demonstrating how to retrieve credentials
from the Secret Store using a credential GUID.
Args:
workflow_args (Dict[str, Any]): The workflow arguments, which
include configuration and a credential GUID reference.
Returns:
Optional[ActivityStatistics]: The activity statistics, or None
if not applicable.
"""
# Step 1: Retrieve the credential GUID from workflow_args.
# This GUID is automatically passed by the Atlan Orchestrator
# during workflow execution, pointing to securely stored secrets.
credential_guid = workflow_args["credential_guid"]
# Step 2: Use SecretStore to resolve the actual credentials.
credentials = await SecretStore.get_credentials(credential_guid)
# Step 3: Credentials are now securely retrieved and can be used
# to establish connections (e.g., database, API).
logger.info("Credentials retrieved successfully")
# Example of how the returned object may look:
# {
# "username": "db_user",
# "password": "secure_password",
# "host": "db.example.com",
# "port": 5432
# }
return None
See also
For more information about credential management and related components:
- Manage credentials - Conceptual overview of credential management
- Credential widget - UI component for credential input