Automate metadata operations

||Connect docs via MCP

Atlan MCP exposes every catalog operation as a callable tool—which means any automation platform can invoke them directly. Use these patterns to run governance sweeps, lineage impact analyses, glossary bootstrapping, stale asset reports, and data quality sweeps from Python, n8n, LangChain, or any agent framework.

Unlike conversational use cases where a person types a prompt, these patterns run programmatically—on a schedule, triggered by pipeline events, or as part of a broader data platform workflow. Each pattern shows the exact tool calls, the order they run in, and ready-to-use code.

info

Make sure the Atlan MCP server is running and your agent is connected before using any of these patterns. For setup, see Set up Atlan MCP.

Governance sweep

Scan for tables that have descriptions but no certification, and automatically apply DRAFT status to flag them for owner review.

How it works

1

Find tables with descriptions and no certification

search_assets

2

Apply certification status in bulk using the GUIDs returned by the search

update_assets

Python

tables = await client.call_tool("search_assets_tool", {
  "asset_type": "Table",
  "conditions": {"user_description": "has_any_value"},
  "negative_conditions": {"certificate_status": "has_any_value"},
  "limit": 50
})

await client.call_tool("update_assets_tool", {
  "assets": [
      {
          "guid": t["guid"],
          "name": t["name"],
          "type_name": "Table",
          "qualified_name": t["qualified_name"]
      }
      for t in tables["results"]
  ],
  "attribute_name": "certificate_status",
  "attribute_values": ["DRAFT"] * len(tables["results"])
})

Lineage impact analysis

Automatically trace which downstream dashboards, models, and datasets are affected when an upstream table changes—before a schema migration or deprecation.

How it works

1

Locate the upstream asset by name

semantic_search

2

Walk downstream dependencies up to the required depth

traverse_lineage

Python

source = await client.call_tool("search_assets_tool", {
  "asset_type": "Table",
  "conditions": {"name": {"operator": "match", "value": "raw_events"}},
  "limit": 1
})

consumers = await client.call_tool("traverse_lineage_tool", {
  "guid": source["results"][0]["guid"],
  "direction": "DOWNSTREAM",
  "depth": 5,
  "size": 100
})

Business graph bootstrap

Programmatically create a glossary, add categories, and populate terms from an external source—each call's output feeds the next, so the order is a genuine dependency chain.

How it works

1

Create the top-level glossary container

create_glossaries

2

Add a category hierarchy using the glossary GUID

3

Populate business terms using both the glossary and category GUIDs

Python

glossary = await client.call_tool("create_glossaries", {
  "name": "Data Dictionary"
})

categories = await client.call_tool("create_glossary_categories", [{
  "name": "Customer",
  "glossary_guid": glossary[0]["guid"]
}])

await client.call_tool("create_glossary_terms", [{
  "name": "Customer ID",
  "glossary_guid": glossary[0]["guid"],
  "category_guids": [categories[0]["guid"]]
}])

Stale asset report

Detect tables without descriptions on a daily schedule and post a summary to Slack so your team can prioritize which assets to document next.

How it works

1

Find tables with missing descriptions

search_assets

2

Format and post results to a Slack channel

n8n

Schedule Trigger (daily 08:00)
→ MCP Client node: Execute Tool — search_assets_tool
    { "asset_type": "Table", "negative_conditions": { "user_description": "has_any_value" }, "limit": 50 }
→ Code node: format asset names into a Slack message body
→ Slack node: post to #data-quality channel

For n8n MCP Client setup, see Set up n8n with Remote MCP.

Data quality sweep

Add null-count DQ rules to all columns in a critical table and schedule them to run nightly alongside your ETL pipeline.

How it works

1

Find columns in the target table

search_assets

2

Define a null-count rule per column

create_dq_rules

3

Configure a recurring cron run for each rule

schedule_dq_rules

Python

columns = await client.call_tool("search_assets_tool", {
  "asset_type": "Column",
  "conditions": {
      "table_qualified_name": {
          "operator": "match",
          "value": "default/snowflake/<CONNECTION_ID>/DB/SCHEMA/ORDERS"
      }
  },
  "limit": 100
})

for col in columns["results"]:
  rule = await client.call_tool("create_dq_rules_tool", {
      "rule_type": "Null Count",
      "asset_qualified_name": col["table_qualified_name"],
      "column_qualified_name": col["qualified_name"],
      "threshold_value": 0,
      "threshold_compare_operator": "EQUAL",
      "alert_priority": "HIGH"
  })

  await client.call_tool("schedule_dq_rules_tool", {
      "rule_id": rule["id"],
      "cron": "0 2 * * *"
  })