Menu
[v2-only] Migrate a v2 connector app from Argo to native orchestration. For v3 migrations, use /upgrade-v3.
Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Baseado na classificação ocupacional SOC
| name | migrate-to-native |
| description | [v2-only] Migrate a v2 connector app from Argo to native orchestration. For v3 migrations, use /upgrade-v3. |
⚠ v2-only skill — NOT for
refactor-v3. This skill targets the Argo → native orchestration migration for v2 connectors. The classes, decorators, and directory layout referenced below (BaseSQLMetadataExtractionApplication,@activity.defn,get_workflow_args,app/activities/metadata_extraction/, etc.) do not exist on therefactor-v3branch — invoking this skill against a v3 codebase will fail. For v3 migrations, use/upgrade-v3instead.
You are helping migrate an Atlan connector app from Argo-based orchestration to native orchestration — where workflows run via Temporal through the Automation Engine instead of Argo Workflows.
Read the full migration guide before starting:
docs/native-migration-guide.md in application-sdkRead the reference implementation:
atlan-redshift-app — the complete reference for all changesThe atlan-application-sdk base classes handle: connection normalization, hybrid credential resolution (inline + Vault), output path computation. Your job is to wire up the app-specific layer on top.
Work through each section below in order. After each section, verify the change compiles or parses correctly before moving on.
Read these files to understand the current state:
app/handlers/{connector}.py — look for get_configmap, test_auth, credential handlingapp/activities/metadata_extraction/{connector}.py — look for get_workflow_args, _set_stateapp/workflows/metadata_extraction/{connector}.py — look for run() return valueapp/templates/ — check what template files exist (if any)pyproject.toml — check current atlan-application-sdk rev/versionIdentify:
get_workflow_args_set_state is overriddenrun() returns anythingIn pyproject.toml, ensure atlan-application-sdk points to a commit that includes:
HandlerInterface._wrap_configmap()BaseSQLMetadataExtractionActivities._set_state()BaseSQLMetadataExtractionWorkflow.run() returning Dict[str, Any]The minimum required commit is 75a48a934bf0b43751a63779879330e30f0028b8.
If the rev is older, update it and run uv lock to regenerate the lockfile.
Create app/templates/atlan-connectors-{connector}.json.
This file defines the Config tab credential form. It must have:
"config" at the top level (no "id" or "name")name, connector, connectorTypehost, portauth-type with enum values matching the connector's auth methodsbasic, iam, role)extra nested object inside each auth type for connector-specific fields (database, deployment_type, etc.)anyOf array to require the correct nested object based on auth-typeUse atlan-redshift-app/app/templates/atlan-connectors-redshift.json as the reference structure.
Create app/templates/workflow.json.
This file defines the Schedule tab workflow form. It must have:
"id": the connector name (e.g. "redshift")"name": display name"logo": URL to connector logoconfig.properties with:
connection — widget "connection"credential-guid — widget "credential", credentialType = "atlan-connectors-{connector}"include-filter / exclude-filter — widget "sqltree" with "sql": "show atlan schemas"preflight-check — widget "sage" with connector-specific checks arrayconfig.steps with at minimum: credential, connection, metadataUse atlan-redshift-app/app/templates/workflow.json as the reference structure.
In app/handlers/{connector}.py, add or replace get_configmap:
@staticmethod
async def get_configmap(config_map_id: str) -> Dict[str, Any]:
base = Path().cwd() / "app" / "templates"
if config_map_id == "atlan-connectors-{connector}":
path = base / "atlan-connectors-{connector}.json"
else:
path = base / "workflow.json"
with open(path) as f:
raw = json.load(f)
return {ClassName}._wrap_configmap(config_map_id, raw)
Add import json and from pathlib import Path if not already present.
Remove any previous get_configmap implementation that read from Elasticsearch or K8s.
In app/activities/metadata_extraction/{connector}.py:
Remove any _set_state override — the SDK handles both native (inline + Vault) and Argo (full Vault fetch) paths automatically.
Replace the full get_workflow_args override with a slim version that:
await super().get_workflow_args(workflow_config)metadata dict)credential is forwarded if present in workflow_config but not yet in workflow_argsExample:
@activity.defn
async def get_workflow_args(self, workflow_config: Dict[str, Any]) -> Dict[str, Any]:
workflow_args = await super().get_workflow_args(workflow_config)
# Connector-specific: flatten metadata keys
metadata = workflow_args.get("metadata") or {}
for key, value in metadata.items():
if key not in workflow_args:
workflow_args[key] = value
if "credential" not in workflow_args and "credential" in workflow_config:
workflow_args["credential"] = workflow_config["credential"]
return workflow_args
Remove all imports that are no longer used after this change (e.g. SecretStore, get_workflow_id, DEPLOYMENT_NAME, build_output_path, etc.).
In app/workflows/metadata_extraction/{connector}.py:
Change await super().run(workflow_config) to output_paths = await super().run(workflow_config) and return output_paths.
Remove any manual output path computation that was previously done in run() (the SDK now computes these from workflow_args["output_path"], workflow_args["output_prefix"], and workflow_args["connection"]).
Remove unused imports (json, Path, etc.) that are no longer needed.
The SDK generates the app manifest automatically via BaseSQLMetadataExtractionApplication.get_manifest(). This produces a two-node DAG — extract (runs the connector's Temporal workflow) followed by publish (runs the Publish App) — covering the full extraction + publish pipeline. No manifest file needs to be created.
Check whether the default manifest covers the connector's needs:
connection, credential-guid, credential, include-filter, exclude-filter, temp-table-regex, extraction-method, advanced-config, etc.) → no change needed.get_manifest() to add them to dag["extract"]["inputs"]["args"].get_manifest().Example override for adding an extra parameter:
class MyConnectorApplication(BaseSQLMetadataExtractionApplication):
def get_manifest(self):
base = super().get_manifest()
base["dag"]["extract"]["inputs"]["args"]["my-extra-param"] = "{{my-extra-param}}"
return base
The SDK's WorkflowInterface.run() base method returns Optional[Dict[str, Any]]. Ensure your workflow's run() return type annotation matches: -> Dict[str, Any] (or Optional[Dict[str, Any]]).
Run the project's type checker if available:
uv run pyright app/
Fix any reported type errors before continuing.
uv run ruff check app/ --fix
uv run ruff format app/
Re-stage any reformatted files before committing.
If local dev infrastructure is available (Dapr + Temporal), run the app and verify:
GET /workflows/v1/configmap/atlan-connectors-{connector} — returns credential form wrapped in { success, message, data: ConfigMap }GET /workflows/v1/configmap/{connector} — returns workflow formPOST /workflows/v1/auth with a test credential — returns { success: true }After completing all steps, report:
_set_state was present and has been removedget_workflow_args{ connection_qualified_name, connection_name }SecretStore.get_secret(credential_guid)transformed_data_prefix, publish_state_prefix, current_state_prefix from output_path{ kind: ConfigMap, apiVersion, metadata, data }{ success, message, data } server envelopeget_configmap() — file path routing by config_map_idget_workflow_args() — connector-specific arg reshaping onlyrun() — return output_paths from super().run()get_manifest() only if the connector needs extra params or a different DAG shapeBump a v3 app to the latest application-sdk and adopt the SDK-native preflight gate safely. The gate runs the app's preflight_check handler as the mandatory first activity of every extraction workflow and always reports the verdict; by default it is soft (a NOT_READY verdict is reported but the run proceeds), and blocking real runs on NOT_READY is a per-app opt-in (preflight_gate_mode = "hard"). Classifies the app's rollout bucket, fixes name collisions, audits the handler's status logic against the new semantics, runs an interactive check-design session with the developer (visualized as a decision tree: which checks block, which are advisory, what is missing), hunts for hidden preflight logic by behavior (not name) and consolidates it into the handler so both surfaces run one implementation, adopts typed check errors, and updates tests. Interactive: every blocking/advisory/consolidation decision belongs to the developer; the skill proposes, never imposes.
Create or audit-and-refresh the canonical capability manifest for application-sdk — a single scannable document listing every public symbol (import path, signature, docstring summary) plus every typed Input/Output contract. Run when starting a new agent task that needs a fast picture of what the SDK exposes, when SDK code has changed, or on a 30-day cadence.
Drive the conformance remediation loop: detect violations, propose and verify fixes, and emit a residue report for anything that needs human review. Runs the conformance suite (deterministic) to detect violations, uses the model to propose fixes, re-runs the suite to verify each fix, and loops until the gate is clean or the attempt cap is reached. Never games its own gate — source fixes are verified by re-detection; logic fixes are also verified by the orthogonal test gate. Backed by the OpenProse program shipped in the atlan-application-sdk-conformance package (resolve it with `programs-dir`). Run with the OpenProse skill to use the full Reactor-ready contract semantics; or invoke the program directly via the instructions below.
Drive an already-open PR to merge-ready: fix CI failures, resolve GitHub Copilot review comments, then loop @sdk-review (mothership) — fixing every finding it reports, including nits — until it comes back clean.
Good investigations depend on good signals. Use this skill to improve the quality of observability in a Python codebase — surfacing failures that are currently hidden, and removing noise that buries the signals you need. Errors swallowed silently, stack traces discarded, log output that can't be queried, credentials leaking into logs, INFO-level chatter drowning out what matters: all of these slow down debugging and erode trust in your observability stack. Standardising log levels means DEBUG stays out of production, INFO marks genuine lifecycle events, and ERROR only fires when something actually failed — so when you filter to WARNING or above during an incident, you see exactly what you need and nothing you don't. Run this before an incident review, before onboarding a service, or any time investigations feel harder than they should.
Convert an app's untyped exceptions (ValueError, RuntimeError, bare Exception, ActivityError) to SDK-typed AppError subclasses so every classified failure carries category / code / audience / retryable on the wire. The Automation Engine attributes failures from those typed fields without parsing exception strings. Interactive and incremental: surveys the app first, asks the developer to confirm categories at ambiguous sites, applies edits one batch at a time, then reports remaining blind-spot (silent-swallow) sites as a phase-2 punch list for owner approval.