diff --git a/.github/workflows/mcp-cli-publish.yaml b/.github/workflows/mcp-cli-publish.yaml new file mode 100644 index 0000000..c5b95eb --- /dev/null +++ b/.github/workflows/mcp-cli-publish.yaml @@ -0,0 +1,141 @@ +name: Publish atlan-cli + +on: + pull_request: + types: [closed] + branches: + - main + workflow_dispatch: + +jobs: + prepare-release: + if: > + github.event_name == 'workflow_dispatch' || + (github.event.pull_request.merged == true && + contains(github.event.pull_request.labels.*.name, 'release-cli')) + runs-on: ubuntu-latest + permissions: + contents: write + outputs: + version: ${{ steps.get_version.outputs.version }} + should_release: ${{ steps.check_tag.outputs.exists == 'false' }} + steps: + - name: Checkout + uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: Get version + id: get_version + run: | + VERSION=$(grep -m 1 "__version__" mcp-cli/atlan_cli.py | cut -d'"' -f2) + echo "version=$VERSION" >> $GITHUB_OUTPUT + echo "Found version: $VERSION" + + - name: Check if tag exists + id: check_tag + run: | + TAG_NAME="mcp-cli-v${{ steps.get_version.outputs.version }}" + if git rev-parse "$TAG_NAME" >/dev/null 2>&1; then + echo "Tag $TAG_NAME already exists, stopping workflow" + echo "exists=true" >> $GITHUB_OUTPUT + else + echo "Tag $TAG_NAME does not exist, continuing" + echo "exists=false" >> $GITHUB_OUTPUT + fi + + - name: Generate changelog + id: changelog + if: steps.check_tag.outputs.exists == 'false' + run: | + set +e + VERSION="${{ steps.get_version.outputs.version }}" + RELEASE_DATE=$(date +"%Y-%m-%d") + PREV_TAG=$(git describe --tags --abbrev=0 HEAD~1 2>/dev/null || echo "") + + if [ -z "$PREV_TAG" ]; then + FIRST_COMMIT=$(git rev-list --max-parents=0 HEAD) + RANGE="$FIRST_COMMIT..HEAD" + else + RANGE="$PREV_TAG..HEAD" + fi + + echo "## [$VERSION] - $RELEASE_DATE" > RELEASE_NOTES.md + echo "" >> RELEASE_NOTES.md + + git log $RANGE --format="* %s (%h)" --grep="^feat" --perl-regexp --no-merges 2>/dev/null > features.txt || touch features.txt + git log $RANGE --format="* %s (%h)" --grep="^fix" --perl-regexp --no-merges 2>/dev/null > fixes.txt || touch fixes.txt + + if [ -s features.txt ]; then + echo "### Added" >> RELEASE_NOTES.md + echo "" >> RELEASE_NOTES.md + sed 's/^\* feat[[:space:]]*\([^:]*\):[[:space:]]*/\* /' features.txt >> RELEASE_NOTES.md + echo "" >> RELEASE_NOTES.md + fi + + if [ -s fixes.txt ]; then + echo "### Fixed" >> RELEASE_NOTES.md + echo "" >> RELEASE_NOTES.md + sed 's/^\* fix[[:space:]]*\([^:]*\):[[:space:]]*/\* /' fixes.txt >> RELEASE_NOTES.md + echo "" >> RELEASE_NOTES.md + fi + + if [ ! -s features.txt ] && [ ! -s fixes.txt ]; then + echo "### Changed" >> RELEASE_NOTES.md + echo "" >> RELEASE_NOTES.md + echo "* Release version $VERSION" >> RELEASE_NOTES.md + echo "" >> RELEASE_NOTES.md + fi + + rm -f features.txt fixes.txt + cat RELEASE_NOTES.md + + - name: Create tag + if: steps.check_tag.outputs.exists == 'false' + run: | + git tag mcp-cli-v${{ steps.get_version.outputs.version }} + git push --tags + + - name: Create GitHub Release + if: steps.check_tag.outputs.exists == 'false' + uses: softprops/action-gh-release@v2 + with: + tag_name: mcp-cli-v${{ steps.get_version.outputs.version }} + name: atlan-cli v${{ steps.get_version.outputs.version }} + body_path: RELEASE_NOTES.md + token: ${{ secrets.GITHUB_TOKEN }} + draft: false + prerelease: false + + - name: Upload release notes + if: steps.check_tag.outputs.exists == 'false' + uses: actions/upload-artifact@v4 + with: + name: release-notes + path: RELEASE_NOTES.md + retention-days: 1 + + publish: + needs: prepare-release + if: needs.prepare-release.outputs.should_release == 'true' + runs-on: ubuntu-latest + permissions: + contents: read + id-token: write # required for PyPI Trusted Publisher (OIDC) + defaults: + run: + working-directory: mcp-cli + steps: + - name: Checkout + uses: actions/checkout@v4 + with: + ref: mcp-cli-v${{ needs.prepare-release.outputs.version }} + + - name: Install uv + uses: astral-sh/setup-uv@v5 + + - name: Build + run: uv build + + - name: Publish to PyPI + run: uv publish diff --git a/mcp-cli/README.md b/mcp-cli/README.md new file mode 100644 index 0000000..cd747af --- /dev/null +++ b/mcp-cli/README.md @@ -0,0 +1,160 @@ +# Atlan MCP CLI + +A standalone CLI for calling Atlan MCP tools directly from your terminal — no IDE, no agent required. Works on macOS, Windows, and Linux. + +## Installation + +Install [`uv`](https://docs.astral.sh/uv/getting-started/installation/) if you don't have it, then install the CLI as a global tool: + +```bash +# From PyPI (once published) +uv tool install atlan-cli + +# From source +uv tool install /path/to/agent-toolkit/mcp-cli +``` + +Add `~/.local/bin` to your PATH if prompted: + +```bash +export PATH="$HOME/.local/bin:$PATH" +``` + +## Quick Start + +```bash +# One-time login — interactive: choose OAuth or API key +atlan login + +# Or skip the prompt and log in directly +atlan login --oauth # browser login +atlan login --api-key sk-xxx --tenant https://your-tenant.atlan.com # API key + +# Check your auth status +atlan status + +# Run tools — no flags needed after login +atlan semantic_search_tool --user-query "PII tables in Snowflake" +atlan list-tools +atlan get_asset_tool --guid "abc-123" +``` + +`atlan login` with no flags shows an interactive prompt: + +``` +Choose login method: + 1 OAuth — browser login via mcp.atlan.com + 2 API key — paste your Atlan API key + +Choice [1/2] (1): 2 + API key: + Tenant URL (e.g. https://demo.atlan.com): https://your-tenant.atlan.com +``` + +## Auth Commands + +| Command | Description | +|---------|-------------| +| `atlan login` | Interactive: choose OAuth browser login or API key | +| `atlan login --oauth` | Force OAuth browser flow directly | +| `atlan login --api-key KEY --tenant URL` | Log in with an Atlan API key | +| `atlan logout` | Remove all stored credentials | +| `atlan status` | Show auth mode, tenant, and token expiry | + +OAuth access tokens auto-refresh via the `mcp.atlan.com` proxy — you rarely need to re-run `atlan login`. To switch tenants, run `atlan logout` then `atlan login` again. + +## Global Flags + +These override stored credentials for a single invocation and are not persisted: + +| Flag | Description | +|------|-------------| +| `--oauth` | Force fresh OAuth browser login this call | +| `--api-key KEY --tenant URL` | One-shot API key (not saved) | +| `--json` | Raw JSON to stdout; all logs go to stderr | + +```bash +# One-shot overrides +atlan --oauth semantic_search_tool --user-query "PII tables" +atlan --api-key sk-xxx --tenant https://demo.atlan.com list-tools +atlan --json get_asset_tool --guid abc-123 +``` + +## Available Tools + +Run `atlan list-tools` to see the full list (requires auth). Common tools by category: + +| Category | Tools | +|----------|-------| +| **Search & discovery** | `semantic_search_tool`, `search_assets_tool`, `search_atlan_docs_tool` | +| **Lineage** | `traverse_lineage_tool` | +| **Asset detail** | `get_asset_tool`, `resolve_metadata_tool`, `get_groups_tool` | +| **Asset updates** | `update_assets_tool`, `manage_announcements_tool`, `manage_asset_lifecycle_tool` | +| **Glossary** | `create_glossaries`, `create_glossary_terms`, `create_glossary_categories` | +| **Data mesh** | `create_domains`, `create_data_products` | +| **Data quality** | `create_dq_rules_tool`, `update_dq_rules_tool`, `schedule_dq_rules_tool`, `delete_dq_rules_tool` | +| **Custom metadata** | `create_custom_metadata_set_tool`, `update_custom_metadata_tool`, `remove_custom_metadata_tool`, `add_attributes_to_cm_set_tool`, `remove_attributes_from_cm_set_tool`, `delete_custom_metadata_set_tool` | +| **Tags** | `add_atlan_tags_tool`, `remove_atlan_tag_tool` | +| **Query** | `query_assets_tool`, `query_deep_sql_tool` | + +Write tools (`create_*`, `update_*`, `delete_*`) default to `--mode propose` which shows a preview without making changes. Pass `--mode execute` only after reviewing the proposal. + +```bash +# Preview a change (safe — no writes) +atlan update_assets_tool --updates '{"guid":"abc","name":"t","qualified_name":"qn","type_name":"Table","user_description":"new desc"}' --mode propose + +# See tool parameters +atlan semantic_search_tool --help +``` + +## resolve_metadata_tool — Valid Namespace Types + +The `--namespace-type` argument accepts these values: + +- `users` — search Atlan users +- `classifications` — Atlan tag/classification names +- `business_metadata` — custom metadata set names +- `glossary` — glossary names and terms +- `data_domain_and_product` — data domains and products + +```bash +atlan resolve_metadata_tool --namespace-type users --query "john" +atlan resolve_metadata_tool --namespace-type glossary --query "revenue" +``` + +## How Credentials Are Stored + +| Storage | Contents | Permissions | +|---------|----------|-------------| +| `~/.atlan/config.json` | Auth mode and tenant URL (no secrets) | 0600 | +| OS keychain (`atlan-mcp`) | Access token, refresh token, or API key | OS-encrypted | +| `~/.atlan/credentials.json` | Fallback when OS keychain is unavailable | 0600 | + +The OS keychain is used automatically on macOS (Keychain), Windows (Credential Manager), and Linux (Secret Service / GNOME Keyring). The file fallback activates on headless servers and CI environments. + +## Exit Codes + +| Code | Meaning | +|------|---------| +| `0` | Success | +| `1` | Tool returned an error | +| `2` | Not authenticated — run `atlan login` | +| `3` | Config or invocation error | + +## Updating + +After pulling new changes: + +```bash +uv tool install /path/to/agent-toolkit/mcp-cli --reinstall +``` + +## Regenerating the CLI + +If tool schemas change (new tools added to the server), regenerate with: + +```bash +fastmcp generate-cli https://your-tenant.atlan.com/mcp --auth oauth --output atlan_cli.py --force +``` + +Re-apply the auth/packaging block at the top after regeneration — `fastmcp` does not write auth into generated scripts. diff --git a/mcp-cli/SKILL.md b/mcp-cli/SKILL.md new file mode 100644 index 0000000..972ef6d --- /dev/null +++ b/mcp-cli/SKILL.md @@ -0,0 +1,504 @@ +--- +name: "mcp.atlan.com-cli" +description: "CLI for the mcp.atlan.com MCP server. Call tools, list resources, and get prompts." +--- + +# mcp.atlan.com CLI + +## Tool Commands + +### semantic_search_tool + +PREFERRED search tool — use this FIRST for any discovery or lookup query. Handles natural language, fuzzy matching, typos, abbreviations, multi-word names, and glossary term lookups. Covers all asset types: tables, columns, views, schemas, dashboards, glossary terms, categories, domains, data products, and more. NOT for users or groups — use resolve_metadata_tool for those. Only fall back to search_assets_tool if you need exact attribute filtering, aggregations, or structured conditions that semantic search cannot express. This tool accepts ONLY: user_query (required), limit, offset, include_readme. It does NOT accept asset_type, conditions, search_query, or any other parameters. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool semantic_search_tool --user-query --limit --offset --include-readme +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--user-query` | string | yes | Natural language search query | +| `--limit` | string | no | Max results to return (default 10, max 20) (JSON string) | +| `--offset` | string | no | Skip first N results for pagination (JSON string) | +| `--include-readme` | boolean | no | Set true to fetch and return README content for each asset. Only enable when the user explicitly asks for README/documentation content — adds latency. | + +### query_deep_sql_tool + +Execute Gold Layer SQL for pagination in the deep query results widget. Internal tool. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool query_deep_sql_tool --sql --limit --offset +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--sql` | string | yes | SELECT SQL query to execute against the Gold Layer. | +| `--limit` | integer | no | Max rows to return. | +| `--offset` | integer | no | Row offset for pagination. | + +### search_assets_tool + +BACKUP search tool — only use when the primary search tools (start_deep_query_tool or semantic_search_tool) cannot express what you need. Provides structured asset search with exact filters, conditions, aggregations, sorting, and pagination. Use this ONLY for precise attribute filters (e.g. certificateStatus, connectorName), term_guids, tags, domain_guids, count_only, or aggregations that the primary tools don't support. + +LINKED/UNLINKED TERMS — MANDATORY 2-CALL FLOW: +When the user asks about linked, unlinked, orphan, or unused glossary terms you MUST make TWO separate calls: +Call 1: Get terms — search_assets(asset_type="AtlasGlossaryTerm", glossary_qualified_name="", include_attributes=["qualifiedName","name"], limit=100) +Call 2: Get linked term QNs — search_assets(aggregations={"linked_terms": {"field": "__meanings", "size": 500}}, limit=1) — NO asset_type, NO glossary filter. This aggregates __meanings across ALL assets in the catalog. +Then DIFF: terms from Call 1 whose qualifiedName appears in Call 2's aggregation buckets are LINKED. Terms absent are UNLINKED. +IMPORTANT: Call 2 must NOT have asset_type or glossary_qualified_name filters — __meanings lives on regular assets (Tables, Columns), not on terms. Scoping Call 2 to terms returns empty buckets. + +GLOSSARY TERM WORKFLOWS: +- Filter terms by category: conditions={"__categories": ""}. NOTE: __categories only stores the DIRECT parent category. To include subcategories, first fetch all categories in the glossary (asset_type="AtlasGlossaryCategory", include_attributes=["qualifiedName","parentCategory"]), walk the parentCategory tree to find all descendants, then use conditions={"__categories": {"operator": "within", "value": [allDescendantQNs]}}. +NOT for users/groups — use resolve_metadata_tool for those. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool search_assets_tool --conditions --negative-conditions --some-conditions --min-somes --include-attributes --asset-type --glossary-qualified-name --include-archived --limit --offset --sort-by --sort-order --sort --connection-qualified-name --tags --directly-tagged --domain-guids --date-range --guids --term-guids --aggregations --count-only --scroll --include-readme --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--conditions` | string | no | Match filters. Format: {attr: value} or {attr: {operator, value}}. For glossary terms in a category: {"__categories": ""} (direct only) or {"__categories": {"operator": "within", "value": [qn1, qn2, ...]}} (multiple categories including subcategories). DO NOT use relationship-type attributes (e.g. meanings, atlanTags, parentCategory, seeAlso) as condition keys — they are not searchable. Use term_guids to filter by linked glossary terms and tags to filter by Atlan tags. (JSON string) | +| `--negative-conditions` | string | no | Exclusion filters (same format as conditions) (JSON string) | +| `--some-conditions` | string | no | OR-style filters requiring min_somes matches (JSON string) | +| `--min-somes` | integer | no | Minimum some_conditions to match | +| `--include-attributes` | string | no | Attributes to return (e.g., owner_users, columns, readme). For category hierarchy traversal include parentCategory and qualifiedName. (JSON string) | +| `--asset-type` | string | no | Asset type(s) to search. Single type or list of types. (JSON string) | +| `--glossary-qualified-name` | string | no | Glossary qualifiedName to scope search to terms/categories within that glossary (JSON string) | +| `--include-archived` | boolean | no | Include archived/deleted assets | +| `--limit` | integer | no | Max results to return (default 10, max 20) | +| `--offset` | integer | no | Pagination offset | +| `--sort-by` | string | no | Single field to sort by (JSON string) | +| `--sort-order` | string | no | Sort order | +| `--sort` | string | no | Multi-field sort (JSON string) | +| `--connection-qualified-name` | string | no | Filter by connection (e.g., default/snowflake/123) (JSON string) | +| `--tags` | string | no | Filter by Atlan tag names (JSON string) | +| `--directly-tagged` | boolean | no | Only directly tagged (not inherited) | +| `--domain-guids` | string | no | Filter by domain GUIDs. For DataProduct/DataDomain asset types, this recursively includes all sub-domains — e.g. passing a parent domain GUID returns products under that domain AND all its sub-domains at any depth. (JSON string) | +| `--date-range` | string | no | Date filters (JSON string) | +| `--guids` | string | no | Filter by specific asset GUIDs (JSON string) | +| `--term-guids` | string | no | Filter by assigned glossary term GUIDs (JSON string) | +| `--aggregations` | string | no | Aggregations to compute. Use __meanings aggregation to find which glossary terms are linked to assets. (JSON string) | +| `--count-only` | boolean | no | Return only count, no results | +| `--scroll` | boolean | no | Enable scroll mode for >10k results | +| `--include-readme` | boolean | no | Set true to fetch and return README content for each asset. Only enable when the user explicitly asks for README/documentation content — adds latency. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### traverse_lineage_tool + +Traverse upstream or downstream lineage from an asset. + +Returns a widget-ready graph with data assets connected by lineage edges. +Process connector nodes are bridged in the graph (A -> B instead of +A -> Process -> B); their entity data is exposed in the ``process_map`` +field and each relation carries ``via_process_guids`` identifying which +ETL/transformation processes produced the connection. The lineage widget +supports progressive loading — leaf nodes show an expand button that +fetches deeper lineage on demand. Keep size small (5-15) for responsive +results; do NOT request all lineage at once. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool traverse_lineage_tool --guid --direction --depth --size --immediate-neighbors --offset --include-attributes --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--guid` | string | yes | GUID of the starting asset | +| `--direction` | string | yes | Lineage traversal direction | +| `--depth` | integer | no | Maximum depth to traverse | +| `--size` | integer | no | Maximum number of data assets to show in the lineage graph. The widget supports progressive loading — users can click expand on leaf nodes to load more. Keep this small (5-15). Max 20. | +| `--immediate-neighbors` | boolean | no | Only return immediate neighbors (one hop) | +| `--offset` | integer | no | Pagination offset for relations (default 0) | +| `--include-attributes` | string | no | Additional attributes to include (JSON string) | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### query_assets_tool + +Execute SQL queries against Atlan connections to preview data. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool query_assets_tool --sql --connection-qualified-name --default-schema --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--sql` | string | yes | SQL query to execute against the connection | +| `--connection-qualified-name` | string | yes | Qualified name of the connection to query | +| `--default-schema` | string | no | Default schema for unqualified table references (JSON string) | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### resolve_metadata_tool + +Use this to search for users and groups, and to get exact usernames and group names before making updates. Also use for extra discovery on classifications, business_metadata, glossary, and data_domain_and_product when semantic_search doesn't return needed results, or before write operations to confirm exact names and GUIDs. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool resolve_metadata_tool --namespace-type --query --limit --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--namespace-type` | string | yes | Metadata namespace to search. Use 'data_domain_and_product' for BOTH data domains AND data products. | +| `--query` | string | yes | Search query - name, description, or natural language | +| `--limit` | integer | no | Max results (default 10, max 20) | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### search_atlan_docs_tool + +Search Atlan's customer-facing documentation and return an LLM-generated answer with source citations. Use for how-to questions about Atlan features — not for searching data assets (use semantic_search_tool for that). + +```bash +uv run --with fastmcp python atlan_cli.py call-tool search_atlan_docs_tool --query --top-k --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--query` | string | yes | Question about Atlan features, how-tos, or configuration. E.g. 'How do I connect Snowflake to Atlan?' or 'How do I create a data product?' | +| `--top-k` | integer | no | Number of documentation sources to retrieve (1-10) | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### get_groups_tool + +Get workspace groups and their members. Use include_members=True to list users in a group. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool get_groups_tool --name-filter --group-id --include-members --limit --offset --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--name-filter` | string | no | Filter by name (partial match) (JSON string) | +| `--group-id` | string | no | Get specific group by ID (GUID) (JSON string) | +| `--include-members` | boolean | no | Include group members in response | +| `--limit` | integer | no | Maximum results (1-250) | +| `--offset` | integer | no | Pagination offset | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### get_asset_tool + +Get detailed information about a single asset by its GUID or qualified name. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool get_asset_tool --guid --qualified-name --asset-type --include-attributes --include-dq-checks --include-readme --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--guid` | string | no | Asset GUID to retrieve (JSON string) | +| `--qualified-name` | string | no | Asset qualified name to retrieve (e.g. default/snowflake/1234/DB/SCHEMA/TABLE) (JSON string) | +| `--asset-type` | string | no | Asset type (required when using qualified_name). e.g. Table, Column, View, Database, Schema (JSON string) | +| `--include-attributes` | string | no | Additional attributes to include (JSON string) | +| `--include-dq-checks` | boolean | no | Set true to include linked data quality checks (Soda, Anomalo, Monte Carlo, Atlan native DQ rules) for this asset | +| `--include-readme` | boolean | no | Set true to fetch and return README content for the asset. Only enable when the user explicitly asks for README/documentation content — adds latency. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### update_assets_tool + +Update attributes on one or more assets. Each item specifies the asset identity + fields to change. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool update_assets_tool --updates --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--updates` | string | yes | Asset update(s). Each update is a self-contained dict with asset identity (guid, name, qualified_name, type_name) plus the fields to change: user_description, certificate_status, readme, term, owner_users, owner_groups, category_guids, term_relations. Multiple attributes can be updated per asset in one call. For owner_users/owner_groups: REPLACES the full list — include existing owners to keep them. Use exact usernames from resolve_metadata. For category_guids: list of category GUIDs to assign an AtlasGlossaryTerm to (replaces existing categories). For AtlasGlossaryTerm or AtlasGlossaryCategory: also include glossary_guid (the parent glossary GUID, available as glossaryGuid in search results). For term: operation must be exactly 'append' (link terms), 'remove' (unlink terms), or 'replace' (overwrite all). Do NOT use 'add' or 'delete' — these are invalid and will fail. For term_relations (AtlasGlossaryTerm only): link one term to another using a relation type. Supported types: synonyms, antonyms, see_also (Related to), preferred_terms (Recommended), preferred_to_terms, translated_terms (Translates to), valid_values, valid_values_for, classifies, is_a (Classified by), replaced_by, replacement_terms. Each relation type takes {op, guids} where op is 'append', 'replace', or 'remove' and guids is a list of target term GUIDs. Use search_assets_tool to find GUIDs of both source and target terms before calling this tool. (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### create_glossaries + +Create one or more glossaries in Atlan. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool create_glossaries --glossaries --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--glossaries` | string | yes | Glossary definitions (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### create_glossary_terms + +Create one or more glossary terms in Atlan. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool create_glossary_terms --terms --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--terms` | string | yes | Term definitions (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### create_glossary_categories + +Create one or more glossary categories in Atlan. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool create_glossary_categories --categories --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--categories` | string | yes | Category definitions (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### create_domains + +Create one or more data domains in Atlan. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool create_domains --domains --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--domains` | string | yes | Domain definitions (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### create_data_products + +Create one or more data products in Atlan. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool create_data_products --products --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--products` | string | yes | Product definitions (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### create_dq_rules_tool + +Create data quality rules in Atlan. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool create_dq_rules_tool --rules --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--rules` | string | yes | DQ rule definitions. Use EXACTLY these field names: rule_type (str, REQUIRED): Null Count, Null Percentage, Blank Count, Blank Percentage, Min Value, Max Value, Average, Standard Deviation, Unique Count, Duplicate Count, Regex, String Length, Valid Values, Valid Values Reference, Freshness, Row Count, Custom SQL, Recon Row Count, Recon Average, Recon Sum, Recon Duplicate Count, Recon Unique Count. asset_qualified_name (str, REQUIRED): qualified name of the table/view. threshold_value (number, REQUIRED): threshold for the rule. threshold_compare_operator (str): EQUAL (EQ), GREATER_THAN_EQUAL (GTE), LESS_THAN_EQUAL (LTE), BETWEEN (BETWEEN), GREATER_THAN (GT), LESS_THAN (LT). alert_priority (str): URGENT, NORMAL, LOW. column_qualified_name (str): required for column-level rules. custom_sql (str): SQL query, required when rule_type is 'Custom SQL'. rule_name (str): display name, required when rule_type is 'Custom SQL'. dimension (str): required when rule_type is 'Custom SQL': COMPLETENESS, TIMELINESS, ACCURACY, CONSISTENCY, UNIQUENESS, VALIDITY, VOLUME. threshold_unit (str): PERCENTAGE, SECONDS, MINUTES, HOURS, DAYS, WEEKS, MONTHS, YEARS, ABSOLUTE. reference_dataset_qualified_name (str): required for Recon and Reference rules. reference_column_qualified_name (str): required for column-level Recon/Reference rules. description (str): optional rule description. (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### schedule_dq_rules_tool + +Schedule data quality rule execution on assets. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool schedule_dq_rules_tool --schedules --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--schedules` | string | yes | Schedule definitions (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### update_dq_rules_tool + +Update existing data quality rules. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool update_dq_rules_tool --rules --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--rules` | string | yes | DQ rule updates. Use EXACTLY these field names: qualified_name (str, REQUIRED): qualified name of the rule to update. rule_type (str, REQUIRED): Null Count, Null Percentage, Blank Count, Blank Percentage, Min Value, Max Value, Average, Standard Deviation, Unique Count, Duplicate Count, Regex, String Length, Valid Values, Valid Values Reference, Freshness, Row Count, Custom SQL, Recon Row Count, Recon Average, Recon Sum, Recon Duplicate Count, Recon Unique Count. asset_qualified_name (str, REQUIRED): qualified name of the table/view. threshold_value (number): new threshold value. threshold_compare_operator (str): EQUAL (EQ), GREATER_THAN_EQUAL (GTE), LESS_THAN_EQUAL (LTE), BETWEEN (BETWEEN), GREATER_THAN (GT), LESS_THAN (LT). threshold_unit (str): PERCENTAGE, SECONDS, MINUTES, HOURS, DAYS, WEEKS, MONTHS, YEARS, ABSOLUTE. alert_priority (str): URGENT, NORMAL, LOW. custom_sql (str): updated SQL query for Custom SQL rules. rule_name (str): updated display name for Custom SQL rules. dimension (str): for Custom SQL rules: COMPLETENESS, TIMELINESS, ACCURACY, CONSISTENCY, UNIQUENESS, VALIDITY, VOLUME. description (str): optional rule description. (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### delete_dq_rules_tool + +Delete data quality rules by their GUIDs. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool delete_dq_rules_tool --rule-guids --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--rule-guids` | string | yes | GUIDs of DQ rules to delete (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### manage_asset_lifecycle_tool + +Manage asset lifecycle: archive, restore, or permanently purge assets. WARNING: PURGE cannot be undone. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool manage_asset_lifecycle_tool --operation --guids --asset-type --qualified-name --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--operation` | string | yes | Lifecycle operation to perform | +| `--guids` | string | no | Asset GUIDs (required for ARCHIVE/PURGE) (JSON string) | +| `--asset-type` | string | no | Asset type (required for RESTORE) (JSON string) | +| `--qualified-name` | string | no | Qualified name (required for RESTORE) (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### manage_announcements_tool + +Add or remove announcements (information, warning, issue) on assets. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool manage_announcements_tool --asset-guids --operation --announcement-type --title --message --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--asset-guids` | string | yes | Asset GUIDs to update (comma-separated or JSON array) | +| `--operation` | string | yes | Announcement operation | +| `--announcement-type` | string | no | Announcement type (required for SET) (JSON string) | +| `--title` | string | no | Announcement title (required for SET) (JSON string) | +| `--message` | string | no | Announcement message (optional) (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### update_custom_metadata_tool + +Update custom metadata on one or more assets. Set replace=True for full replacement, False (default) for partial update. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool update_custom_metadata_tool --updates --replace --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--updates` | string | yes | Single update (dict) or batch updates (list of dicts). Each item requires: guid, custom_metadata_name, attributes. Single: {"guid": "...", "custom_metadata_name": "CM", "attributes": {"k": "v"}}. Batch: [{"guid": "g1", ...}, {"guid": "g2", ...}] (JSON string) | +| `--replace` | boolean | no | If False (default): partial update — only specified attributes are changed. If True: full replacement — ALL attributes in the CM set are replaced; unspecified attributes are cleared. replace=True only supported for single asset (dict input), not batch. | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### remove_custom_metadata_tool + +Remove a custom metadata set from an asset. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool remove_custom_metadata_tool --guid --custom-metadata-name --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--guid` | string | yes | GUID of the asset | +| `--custom-metadata-name` | string | yes | Name of the custom metadata set to remove | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### create_custom_metadata_set_tool + +Create one or more custom metadata sets with typed attributes. Defines the schema; use update_custom_metadata to set values on assets. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool create_custom_metadata_set_tool --sets --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--sets` | string | yes | Single CM set (dict) or multiple CM sets (list of dicts). Each requires: display_name (str), attributes (list). Optional: description (str). Each attribute requires: display_name, attribute_type (string\|int\|float\|boolean\|date\|enum\|users\|groups\|url\|SQL\|long). Optional per attribute: multi_valued (bool), description (str). For enum attributes: provide enum_values (list of allowed strings) to create a new enum, OR options_name (str) to reference an existing Atlan enum. Single: {"display_name": "Data Quality", "attributes": [{"display_name": "Score", "attribute_type": "int"}, {"display_name": "Dimension", "attribute_type": "enum", "enum_values": ["Accuracy", "Completeness"]}]}. Batch: [{"display_name": "CM1", "attributes": [...]}, {"display_name": "CM2", "attributes": [...]}] (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### delete_custom_metadata_set_tool + +Permanently delete one or more custom metadata sets and all their attribute values from assets. Irreversible. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool delete_custom_metadata_set_tool --sets --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--sets` | string | yes | Single CM set display name (string) or list of display names for batch deletion. WARNING: Irreversible — deletes the schema and all stored attribute values on assets. Single: "Data Quality". Batch: ["Data Quality", "Governance"] (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### add_attributes_to_cm_set_tool + +Add new typed attributes to an existing custom metadata set. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool add_attributes_to_cm_set_tool --display-name --attributes --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--display-name` | string | yes | Display name of the existing CM set | +| `--attributes` | string | yes | Attributes to add. Each requires: display_name, attribute_type (string\|int\|float\|boolean\|date\|enum\|users\|groups\|url\|SQL\|long). Optional: multi_valued (bool), description (str). For enum attributes: provide enum_values (list of allowed strings) to create a new enum, OR options_name (str) to reference an existing Atlan enum. Example: [{"display_name": "Risk Score", "attribute_type": "int"}, {"display_name": "Status", "attribute_type": "enum", "enum_values": ["Draft", "Approved", "Rejected"]}] (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### remove_attributes_from_cm_set_tool + +Remove (archive) attributes from an existing custom metadata set. Archived attributes are soft-deleted and their values cleared from all assets. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool remove_attributes_from_cm_set_tool --display-name --attribute-names --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--display-name` | string | yes | Display name of the existing CM set | +| `--attribute-names` | string | yes | Display name(s) of attributes to remove (archive). Archived attributes are soft-deleted: values are cleared from assets and hidden in UI. Example: ["Score", "Reviewed By"] or a JSON string of the list. (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### add_atlan_tags_tool + +Add Atlan tags to one or more assets. Pass a dict for single asset, list for batch. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool add_atlan_tags_tool --updates --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--updates` | string | yes | Single tag addition (dict) or batch (list of dicts). Each item: {guid, tag_names, propagate (default true), remove_on_delete (default true), restrict_lineage_propagation (default false), restrict_hierarchy_propagation (default false)}. Single: {"guid": "...", "tag_names": ["PII"]}. Batch: [{"guid": "g1", "tag_names": ["PII"]}, ...] (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### remove_atlan_tag_tool + +Remove an Atlan tag from one or more assets. Pass a dict for single asset, list for batch. Always uses propose mode — STOP after proposing and wait for user approval before executing. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool remove_atlan_tag_tool --updates --mode --user-query +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--updates` | string | yes | Single tag removal (dict) or batch (list of dicts). Each item: {guid, tag_name}. Single: {"guid": "...", "tag_name": "PII"}. Batch: [{"guid": "g1", "tag_name": "PII"}, ...] (JSON string) | +| `--mode` | string | no | ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved. | +| `--user-query` | string | no | REQUIRED: Always pass the user's exact question/prompt that triggered this tool call. Used for tracing and observability. (JSON string) | + +### get_asset_icons + +Fetch SVG icons for asset types. Internal tool used by widgets. + +```bash +uv run --with fastmcp python atlan_cli.py call-tool get_asset_icons --icon-names +``` + +| Flag | Type | Required | Description | +|------|------|----------|-------------| +| `--icon-names` | array[string] | yes | | + +## Utility Commands + +```bash +uv run --with fastmcp python atlan_cli.py list-tools +uv run --with fastmcp python atlan_cli.py list-resources +uv run --with fastmcp python atlan_cli.py read-resource +uv run --with fastmcp python atlan_cli.py list-prompts +uv run --with fastmcp python atlan_cli.py get-prompt [key=value ...] +``` diff --git a/mcp-cli/atlan_cli.py b/mcp-cli/atlan_cli.py new file mode 100755 index 0000000..6ea149a --- /dev/null +++ b/mcp-cli/atlan_cli.py @@ -0,0 +1,1959 @@ +#!/usr/bin/env python3 +"""CLI for Atlan MCP server. + +Generated by: fastmcp generate-cli https://your-tenant.atlan.com/mcp +""" + +__version__ = "0.1.0" + +import hashlib +import json +import os +import sys +import time +from pathlib import Path +from typing import Annotated, Any, Mapping + +import cyclopts +import mcp.types +from dotenv import load_dotenv +from fastmcp import Client +from fastmcp.client.auth import BearerAuth, OAuth +from rich.console import Console +from rich.panel import Panel + +load_dotenv(Path.cwd() / ".env", override=False) +load_dotenv(Path(__file__).parent / ".env", override=False) + +_ATLAN_DIR = Path.home() / ".atlan" +_CONFIG_FILE = _ATLAN_DIR / "config.json" +_CREDS_FILE = _ATLAN_DIR / "credentials.json" +_OAUTH_PROXY = "https://mcp.atlan.com" +_JSON_MODE = False + + +class _JsonFileStore: + """Minimal AsyncKeyValue store that persists tokens as JSON files in a directory. + + Avoids FileTreeStore's limitation of using keys as filesystem paths, which + breaks when keys are URLs containing slashes. + """ + + def __init__(self, path: Path) -> None: + self._path = path + self._path.mkdir(parents=True, exist_ok=True) + + def _file(self, collection: str) -> Path: + safe = hashlib.sha256(collection.encode()).hexdigest()[:16] + return self._path / f"{safe}.json" + + def _load(self, collection: str) -> dict: + f = self._file(collection) + return json.loads(f.read_text()) if f.exists() else {} + + def _save(self, collection: str, data: dict) -> None: + self._file(collection).write_text(json.dumps(data)) + + async def get( + self, key: str, *, collection: str | None = None + ) -> dict[str, Any] | None: + return self._load(collection or "default").get(key) + + async def ttl( + self, key: str, *, collection: str | None = None + ) -> tuple[dict[str, Any] | None, float | None]: + return self._load(collection or "default").get(key), None + + async def put( + self, + key: str, + value: Mapping[str, Any], + *, + collection: str | None = None, + ttl: Any = None, + ) -> None: + data = self._load(collection or "default") + data[key] = dict(value) + self._save(collection or "default", data) + + async def delete(self, key: str, *, collection: str | None = None) -> bool: + data = self._load(collection or "default") + if key in data: + del data[key] + self._save(collection or "default", data) + return True + return False + + async def get_many( + self, keys: list[str], *, collection: str | None = None + ) -> list[dict[str, Any] | None]: + data = self._load(collection or "default") + return [data.get(k) for k in keys] + + +def _keyring_get(account: str) -> str | None: + try: + import keyring as kr + + val = kr.get_password("atlan-mcp", account) + if val is not None: + return val + except Exception: + pass + if _CREDS_FILE.exists(): + try: + return json.loads(_CREDS_FILE.read_text()).get(account) + except Exception: + pass + return None + + +def _keyring_set(account: str, value: str) -> None: + try: + import keyring as kr + + kr.set_password("atlan-mcp", account, value) + return + except Exception: + pass + _ATLAN_DIR.mkdir(parents=True, exist_ok=True) + data: dict = {} + if _CREDS_FILE.exists(): + try: + data = json.loads(_CREDS_FILE.read_text()) + except Exception: + pass + data[account] = value + _CREDS_FILE.write_text(json.dumps(data)) + _CREDS_FILE.chmod(0o600) + + +def _keyring_delete(account: str) -> None: + try: + import keyring as kr + + kr.delete_password("atlan-mcp", account) + except Exception: + pass + if _CREDS_FILE.exists(): + try: + data = json.loads(_CREDS_FILE.read_text()) + data.pop(account, None) + _CREDS_FILE.write_text(json.dumps(data)) + except Exception: + pass + + +def _maybe_json(v): + """Parse v as JSON if it's a string; return v unchanged on failure.""" + if not isinstance(v, str): + return v + try: + return json.loads(v) + except (json.JSONDecodeError, ValueError): + return v + + +def _read_config() -> dict | None: + if not _CONFIG_FILE.exists(): + return None + try: + return json.loads(_CONFIG_FILE.read_text()) + except Exception: + _stderr.print( + "[bold red]Error:[/bold red] Malformed ~/.atlan/config.json. Run [cyan]atlan login[/cyan]." + ) + sys.exit(3) + + +def _write_config(cfg: dict) -> None: + _ATLAN_DIR.mkdir(parents=True, exist_ok=True) + _CONFIG_FILE.write_text(json.dumps(cfg, indent=2)) + _CONFIG_FILE.chmod(0o600) + + +def _wipe_credentials() -> None: + """Wipe all stored credentials. Called ONLY from: refresh failure, logout, login.""" + global _resolved + _resolved = None + for account in ("refresh_token", "access_token", "api_key"): + _keyring_delete(account) + # Clear FastMCP's internal token cache so next login --oauth always opens browser. + import shutil + + cache = _ATLAN_DIR / ".fastmcp-cache" + if cache.exists(): + shutil.rmtree(cache) + + +_resolved: tuple[str, object] | None = None + + +def _resolve_auth() -> tuple[str, object]: + global _resolved + if _resolved is not None: + return _resolved + + override_api_key = os.environ.get("_ATLAN_OVERRIDE_API_KEY") + override_tenant = os.environ.get("_ATLAN_OVERRIDE_TENANT", "").rstrip("/") + override_oauth = os.environ.get("_ATLAN_OVERRIDE_OAUTH") == "1" + + if override_api_key and override_tenant: + _resolved = (f"{override_tenant}/mcp/api-key", BearerAuth(override_api_key)) + return _resolved + + if override_oauth: + url = f"{_OAUTH_PROXY}/mcp" + store = _JsonFileStore(_ATLAN_DIR / ".fastmcp-cache") + _resolved = (url, OAuth(mcp_url=url, token_storage=store)) + return _resolved + + cfg = _read_config() + + if cfg is None: + # Legacy env-var fallback for users who haven't run atlan login yet + base_url = os.environ.get("ATLAN_BASE_URL", "").rstrip("/") + if not base_url: + _err("Not authenticated. Run [cyan]atlan login[/cyan].") + sys.exit(2) + api_key = os.environ.get("ATLAN_API_KEY") + force_oauth = os.environ.get("ATLAN_AUTH", "").lower() == "oauth" + if api_key and not force_oauth: + _resolved = (f"{base_url}/mcp/api-key", BearerAuth(api_key)) + else: + store = _JsonFileStore(_ATLAN_DIR / ".fastmcp-cache") + url = f"{base_url}/mcp" + _resolved = (url, OAuth(mcp_url=url, token_storage=store)) + return _resolved + + auth_mode = cfg.get("auth_mode") + + if auth_mode == "api-key": + tenant = cfg.get("tenant", "").rstrip("/") + if not tenant: + _err( + "Config error: api-key mode requires tenant URL. Run [cyan]atlan login[/cyan]." + ) + sys.exit(3) + api_key = _keyring_get("api_key") + if not api_key: + _err("Not authenticated. Run [cyan]atlan login[/cyan].") + sys.exit(2) + _resolved = (f"{tenant}/mcp/api-key", BearerAuth(api_key)) + return _resolved + + if auth_mode == "oauth": + at_raw = _keyring_get("access_token") + if at_raw: + try: + at_data = json.loads(at_raw) + if time.time() < at_data.get("expires_at", 0): + _resolved = ( + f"{_OAUTH_PROXY}/mcp", + BearerAuth(at_data["access_token"]), + ) + return _resolved + except Exception: + pass + + refresh_token = _keyring_get("refresh_token") + if not refresh_token: + _err("Not authenticated. Run [cyan]atlan login[/cyan].") + sys.exit(2) + + import httpx + + try: + resp = httpx.post( + f"{_OAUTH_PROXY}/oauth/token", + data={ + "grant_type": "refresh_token", + "refresh_token": refresh_token, + "client_id": cfg.get("client_id", "mcp-client"), + "client_secret": "placeholder", + }, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + timeout=10, + ) + except Exception as exc: + _err(f"Network error during token refresh: {exc}") + sys.exit(2) + + if resp.status_code != 200: + _wipe_credentials() + _err("Session expired. Run [cyan]atlan login[/cyan].") + sys.exit(2) + + data = resp.json() + access_token = data["access_token"] + expires_in = data.get("expires_in", 300) + _keyring_set( + "access_token", + json.dumps( + { + "access_token": access_token, + "expires_at": time.time() + expires_in - 30, + } + ), + ) + if "refresh_token" in data: + _keyring_set("refresh_token", data["refresh_token"]) + + _resolved = (f"{_OAUTH_PROXY}/mcp", BearerAuth(access_token)) + return _resolved + + _err(f"Unknown auth_mode {auth_mode!r}. Run [cyan]atlan login[/cyan].") + sys.exit(3) + + +def _client() -> "Client": + url, auth = _resolve_auth() + return Client(url, auth=auth) + + +app = cyclopts.App(name="atlan", help="CLI for Atlan MCP server") + +console = Console() +_stderr = Console(stderr=True) + + +def _err(msg: str) -> None: + _stderr.print(msg) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _print_tool_result(result) -> None: + if result.is_error: + for block in result.content: + if isinstance(block, mcp.types.TextContent): + if _JSON_MODE: + print(json.dumps({"error": block.text})) + else: + _stderr.print(f"[bold red]Error:[/bold red] {block.text}") + else: + if _JSON_MODE: + print(json.dumps({"error": str(block)})) + else: + _stderr.print(f"[bold red]Error:[/bold red] {block}") + sys.exit(1) + + if result.structured_content is not None: + if _JSON_MODE: + print(json.dumps(result.structured_content)) + else: + console.print_json(json.dumps(result.structured_content)) + return + + for block in result.content: + if isinstance(block, mcp.types.TextContent): + if _JSON_MODE: + print(block.text) + else: + console.print(block.text) + elif isinstance(block, mcp.types.ImageContent): + size = len(block.data) * 3 // 4 + if not _JSON_MODE: + console.print(f"[dim][Image: {block.mimeType}, ~{size} bytes][/dim]") + elif isinstance(block, mcp.types.AudioContent): + size = len(block.data) * 3 // 4 + if not _JSON_MODE: + console.print(f"[dim][Audio: {block.mimeType}, ~{size} bytes][/dim]") + + +async def _call_tool(tool_name: str, arguments: dict) -> None: + # Filter out None values and empty lists (defaults for optional array params) + filtered = { + k: v + for k, v in arguments.items() + if v is not None and (not isinstance(v, list) or len(v) > 0) + } + async with _client() as client: + result = await client.call_tool(tool_name, filtered, raise_on_error=False) + _print_tool_result(result) + if result.is_error: + sys.exit(1) + + +# --------------------------------------------------------------------------- +# List / read commands +# --------------------------------------------------------------------------- + + +@app.command +async def list_tools() -> None: + """List available tools.""" + from rich.table import Table + + async with _client() as client: + tools = await client.list_tools() + if not tools: + console.print("[dim]No tools found.[/dim]") + return + table = Table(show_header=True, header_style="bold cyan", box=None, padding=(0, 2)) + table.add_column("Tool", style="cyan", no_wrap=True) + table.add_column("Description") + table.add_column("Parameters", style="dim") + for tool in tools: + props = tool.inputSchema.get("properties", {}) + required = set(tool.inputSchema.get("required", [])) + params = ", ".join(f"[bold]{n}[/bold]" if n in required else n for n in props) + table.add_row(tool.name, tool.description or "", params) + console.print(table) + console.print(f"\n[dim]Bold = required · {len(tools)} tools total[/dim]") + + +@app.command +async def list_resources() -> None: + """List available resources.""" + async with _client() as client: + resources = await client.list_resources() + if not resources: + console.print("[dim]No resources found.[/dim]") + return + for r in resources: + console.print(f" [cyan]{r.uri}[/cyan]") + desc_parts = [r.name or "", r.description or ""] + desc = " — ".join(p for p in desc_parts if p) + if desc: + console.print(f" {desc}") + console.print() + + +@app.command +async def read_resource( + uri: Annotated[str, cyclopts.Parameter(help="Resource URI")], +) -> None: + """Read a resource by URI.""" + async with _client() as client: + contents = await client.read_resource(uri) + for block in contents: + if isinstance(block, mcp.types.TextResourceContents): + console.print(block.text) + elif isinstance(block, mcp.types.BlobResourceContents): + size = len(block.blob) * 3 // 4 + console.print(f"[dim][Blob: {block.mimeType}, ~{size} bytes][/dim]") + + +@app.command +async def list_prompts() -> None: + """List available prompts.""" + async with _client() as client: + prompts = await client.list_prompts() + if not prompts: + console.print("[dim]No prompts found.[/dim]") + return + for p in prompts: + args_str = "" + if p.arguments: + parts = [a.name for a in p.arguments] + args_str = f"({', '.join(parts)})" + console.print(f" [cyan]{p.name}{args_str}[/cyan]") + if p.description: + console.print(f" {p.description}") + console.print() + + +@app.command +async def get_prompt( + name: Annotated[str, cyclopts.Parameter(help="Prompt name")], + *arguments: str, +) -> None: + """Get a prompt by name. Pass arguments as key=value pairs.""" + parsed: dict[str, str] = {} + for arg in arguments: + if "=" not in arg: + console.print( + f"[bold red]Error:[/bold red] Invalid argument {arg!r} — expected key=value" + ) + sys.exit(1) + key, value = arg.split("=", 1) + parsed[key] = value + + async with _client() as client: + result = await client.get_prompt(name, parsed or None) + for msg in result.messages: + console.print(f"[bold]{msg.role}:[/bold]") + if isinstance(msg.content, mcp.types.TextContent): + console.print(f" {msg.content.text}") + elif isinstance(msg.content, mcp.types.ImageContent): + size = len(msg.content.data) * 3 // 4 + console.print( + f" [dim][Image: {msg.content.mimeType}, ~{size} bytes][/dim]" + ) + else: + console.print(f" {msg.content}") + console.print() + + +# --------------------------------------------------------------------------- +# Auth commands +# --------------------------------------------------------------------------- + + +@app.command +async def login( + *, + oauth: Annotated[ + bool, cyclopts.Parameter("--oauth", help="Force OAuth browser login") + ] = False, + api_key: Annotated[ + str | None, cyclopts.Parameter("--api-key", help="Atlan API key") + ] = None, + tenant: Annotated[ + str | None, + cyclopts.Parameter("--tenant", help="Tenant URL (required with --api-key)"), + ] = None, +) -> None: + """Authenticate with Atlan. Run once; tool calls reuse stored credentials. + + With no flags: interactive chooser (OAuth vs API key). + --oauth: browser OAuth flow directly. + --api-key KEY --tenant URL: API key mode. + """ + # Always wipe first to prevent mode-switch cross-contamination. + _wipe_credentials() + if _CONFIG_FILE.exists(): + _CONFIG_FILE.unlink() + + if api_key and not tenant: + _err("[bold red]Error:[/bold red] --tenant is required with --api-key") + sys.exit(3) + + if api_key and tenant: + tenant = tenant.rstrip("/") + import httpx + + try: + resp = httpx.get( + f"{tenant}/api/meta/types/typedefs", + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + timeout=10, + follow_redirects=True, + ) + except Exception as exc: + _err(f"Network error validating API key: {exc}") + sys.exit(3) + if resp.status_code in (401, 403): + _err( + f"API key validation failed (HTTP {resp.status_code}). Check key and tenant URL." + ) + sys.exit(3) + _keyring_set("api_key", api_key) + _write_config({"auth_mode": "api-key", "tenant": tenant}) + console.print( + Panel( + f" Mode [cyan]api-key[/cyan]\n Tenant {tenant}\n\nRun [bold]atlan status[/bold] to verify connectivity.", + title="[green]● Login successful[/green]", + expand=False, + ) + ) + return + + use_oauth = oauth or bool(os.environ.get("_ATLAN_OVERRIDE_OAUTH")) + + if not use_oauth: + from rich.prompt import Prompt + + console.print("\n[bold]Choose login method:[/bold]") + console.print(" [cyan]1[/cyan] OAuth — browser login via mcp.atlan.com") + console.print(" [cyan]2[/cyan] API key — paste your Atlan API key\n") + try: + choice = Prompt.ask("Choice", choices=["1", "2"], default="1") + except (KeyboardInterrupt, EOFError): + sys.exit(0) + if choice == "2": + try: + # Use input() not getpass — getpass reads char-by-char in raw mode + # and stalls on long pasted JWTs (1000+ chars). + api_key_in = input(" API key: ").strip() + tenant_in = input( + " Tenant URL (e.g. https://demo.atlan.com): " + ).strip() + except (KeyboardInterrupt, EOFError): + sys.exit(0) + if not api_key_in or not tenant_in: + _err("API key and tenant are required.") + sys.exit(3) + await login(oauth=False, api_key=api_key_in, tenant=tenant_in) + return + use_oauth = True + + # OAuth browser flow. + # FastMCP wraps token_storage in PydanticAdapter, so we can't intercept put() reliably. + # Instead: complete the flow, then read tokens back from the OAuth adapter directly. + url = f"{_OAUTH_PROXY}/mcp" + store = _JsonFileStore(_ATLAN_DIR / ".fastmcp-cache") + auth = OAuth(mcp_url=url, token_storage=store) + console.print(f"Opening browser for OAuth login at [cyan]{_OAUTH_PROXY}[/cyan] …") + try: + async with Client(url, auth=auth) as client: + await client.list_tools() + except Exception as exc: + _err(f"OAuth login failed: {exc}") + sys.exit(2) + + # Extract tokens from the completed OAuth flow and persist to our canonical keyring. + tokens = await auth.token_storage_adapter.get_tokens() + if tokens and tokens.access_token: + expires_at = time.time() + (tokens.expires_in or 300) - 30 + _keyring_set( + "access_token", + json.dumps( + { + "access_token": tokens.access_token, + "expires_at": expires_at, + } + ), + ) + if tokens.refresh_token: + _keyring_set("refresh_token", tokens.refresh_token) + + _write_config({"auth_mode": "oauth", "client_id": "mcp-client"}) + console.print( + Panel( + f" Mode [cyan]oauth[/cyan]\n Proxy {_OAUTH_PROXY}/mcp\n\nRun [bold]atlan status[/bold] to check token validity.", + title="[green]● Login successful[/green]", + expand=False, + ) + ) + + +@app.command +async def logout() -> None: + """Remove stored credentials. Run atlan login to re-authenticate.""" + _wipe_credentials() + if _CONFIG_FILE.exists(): + _CONFIG_FILE.unlink() + console.print( + "[green]Logged out.[/green] Run [cyan]atlan login[/cyan] to authenticate again." + ) + + +@app.command +async def status() -> None: + """Show current authentication state and token validity.""" + cfg = _read_config() + if cfg is None: + _err("Not authenticated. Run [cyan]atlan login[/cyan].") + sys.exit(2) + + auth_mode = cfg.get("auth_mode", "unknown") + + if auth_mode == "api-key": + tenant = cfg.get("tenant", "?") + has_key = bool(_keyring_get("api_key")) + if has_key: + lines = [ + " Mode [cyan]api-key[/cyan]", + f" Tenant {tenant}", + ] + console.print( + Panel( + "\n".join(lines), + title="[green]● Authenticated[/green]", + expand=False, + ) + ) + else: + _err("Config present but API key missing. Run [cyan]atlan login[/cyan].") + sys.exit(2) + + elif auth_mode == "oauth": + at_raw = _keyring_get("access_token") + rt = _keyring_get("refresh_token") + if at_raw: + try: + at_data = json.loads(at_raw) + remaining = at_data.get("expires_at", 0) - time.time() + if remaining > 0: + lines = [ + " Mode [cyan]oauth[/cyan]", + f" Proxy {_OAUTH_PROXY}/mcp", + f" Token expires in [cyan]{int(remaining)}s[/cyan]", + f" Refresh {'[green]present[/green]' if rt else '[red]missing[/red]'}", + ] + console.print( + Panel( + "\n".join(lines), + title="[green]● Authenticated[/green]", + expand=False, + ) + ) + return + except Exception: + pass + if rt: + console.print( + Panel( + " Mode [cyan]oauth[/cyan]\n Token [yellow]expired[/yellow] — will auto-refresh on next call\n Refresh [green]present[/green]", + title="[yellow]● Token Expired[/yellow]", + expand=False, + ) + ) + else: + _err("Not authenticated. Run [cyan]atlan login[/cyan].") + sys.exit(2) + else: + _err(f"Unknown auth_mode {auth_mode!r}. Run [cyan]atlan login[/cyan].") + sys.exit(3) + + +# --------------------------------------------------------------------------- +# Tool commands (generated from server schema) +# --------------------------------------------------------------------------- + + +@app.command(name="semantic_search_tool") +async def semantic_search_tool( + *, + user_query: Annotated[ + str, cyclopts.Parameter(help="Natural language search query") + ], + limit: Annotated[ + str | None, + cyclopts.Parameter( + help='Max results to return (default 10, max 20)\\nJSON Schema: {\n "anyOf": [\n {\n "type": "integer"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Max results to return (default 10, max 20)"\n }' + ), + ] = None, + offset: Annotated[ + str | None, + cyclopts.Parameter( + help='Skip first N results for pagination\\nJSON Schema: {\n "anyOf": [\n {\n "type": "integer"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Skip first N results for pagination"\n }' + ), + ] = None, + include_readme: Annotated[ + bool, + cyclopts.Parameter( + help="Set true to fetch and return README content for each asset. Only enable when the user explicitly asks for README/documentation content — adds latency." + ), + ] = False, +) -> None: + """PREFERRED search tool — use this FIRST for any discovery or lookup query. Handles natural language, fuzzy matching, typos, abbreviations, multi-word names, and glossary term lookups. Covers all asset types: tables, columns, views, schemas, dashboards, glossary terms, categories, domains, data products, and more. NOT for users or groups — use resolve_metadata_tool for those. Only fall back to search_assets_tool if you need exact attribute filtering, aggregations, or structured conditions that semantic search cannot express. This tool accepts ONLY: user_query (required), limit, offset, include_readme. It does NOT accept asset_type, conditions, search_query, or any other parameters.""" + limit_parsed = _maybe_json(limit) + offset_parsed = _maybe_json(offset) + + await _call_tool( + "semantic_search_tool", + { + "user_query": user_query, + "limit": limit_parsed, + "offset": offset_parsed, + "include_readme": include_readme, + }, + ) + + +@app.command(name="query_deep_sql_tool") +async def query_deep_sql_tool( + *, + sql: Annotated[ + str, + cyclopts.Parameter(help="SELECT SQL query to execute against the Gold Layer."), + ], + limit: Annotated[int, cyclopts.Parameter(help="Max rows to return.")] = 50, + offset: Annotated[int, cyclopts.Parameter(help="Row offset for pagination.")] = 0, +) -> None: + """Execute Gold Layer SQL for pagination in the deep query results widget. Internal tool.""" + await _call_tool( + "query_deep_sql_tool", {"sql": sql, "limit": limit, "offset": offset} + ) + + +@app.command(name="search_assets_tool") +async def search_assets_tool( + *, + conditions: Annotated[ + str | None, + cyclopts.Parameter( + help='Match filters. Format: {attr: value} or {attr: {operator, value}}. For glossary terms in a category: {"__categories": ""} (direct only) or {"__categories": {"operator": "within", "value": [qn1, qn2, ...]}} (multiple categories including subcategories). DO NOT use relationship-type attributes (e.g. meanings, atlanTags, parentCategory, seeAlso) as condition keys — they are not searchable. Use term_guids to filter by linked glossary terms and tags to filter by Atlan tags.\\nJSON Schema: {\n "anyOf": [\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Match filters. Format: {attr: value} or {attr: {operator, value}}. For glossary terms in a category: {\\"__categories\\": \\"\\"} (direct only) or {\\"__categories\\": {\\"operator\\": \\"within\\", \\"value\\": [qn1, qn2, ...]}} (multiple categories including subcategories). DO NOT use relationship-type attributes (e.g. meanings, atlanTags, parentCategory, seeAlso) as condition keys — they are not searchable. Use term_guids to filter by linked glossary terms and tags to filter by Atlan tags.",\n "examples": [\n {\n "name": "customers"\n },\n {\n "name": {\n "operator": "startswith",\n "value": "dim_"\n }\n },\n {\n "certificate_status": "VERIFIED"\n },\n {\n "__typeName": "Table"\n },\n {\n "__categories": ""\n },\n {\n "__categories": {\n "operator": "within",\n "value": [\n "",\n "",\n ""\n ]\n }\n }\n ]\n }' + ), + ] = None, + negative_conditions: Annotated[ + str | None, + cyclopts.Parameter( + help='Exclusion filters (same format as conditions)\\nJSON Schema: {\n "anyOf": [\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Exclusion filters (same format as conditions)",\n "examples": [\n {\n "certificate_status": "DEPRECATED"\n },\n {\n "name": {\n "operator": "contains",\n "value": "test"\n }\n }\n ]\n }' + ), + ] = None, + some_conditions: Annotated[ + str | None, + cyclopts.Parameter( + help='OR-style filters requiring min_somes matches\\nJSON Schema: {\n "anyOf": [\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "OR-style filters requiring min_somes matches",\n "examples": [\n {\n "owner_groups": "admins",\n "owner_users": "user1"\n }\n ]\n }' + ), + ] = None, + min_somes: Annotated[ + int, cyclopts.Parameter(help="Minimum some_conditions to match") + ] = 1, + include_attributes: Annotated[ + str | None, + cyclopts.Parameter( + help='Attributes to return (e.g., owner_users, columns, readme). For category hierarchy traversal include parentCategory and qualifiedName.\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "type": "string"\n },\n "type": "array"\n },\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Attributes to return (e.g., owner_users, columns, readme). For category hierarchy traversal include parentCategory and qualifiedName."\n }' + ), + ] = None, + asset_type: Annotated[ + str | None, + cyclopts.Parameter( + help='Asset type(s) to search. Single type or list of types.\\nJSON Schema: {\n "anyOf": [\n {\n "enum": [\n "Table",\n "Column",\n "View",\n "Database",\n "Schema",\n "MaterializedView",\n "AtlasGlossary",\n "AtlasGlossaryTerm",\n "AtlasGlossaryCategory",\n "Connection",\n "Process",\n "Query",\n "Dashboard",\n "Report",\n "DataDomain",\n "DataProduct",\n "PowerBIReport",\n "PowerBIDataset",\n "PowerBIDashboard",\n "PowerBIWorkspace",\n "PowerBIDataflow",\n "PowerBITable",\n "PowerBIMeasure",\n "PowerBIColumn",\n "PowerBIPage",\n "DbtModel",\n "DbtTest",\n "DbtSource",\n "DbtMetric",\n "DbtModelColumn",\n "TableauDashboard",\n "TableauCalculatedField",\n "TableauWorkbook",\n "TableauDatasource",\n "TableauProject",\n "TableauWorksheet",\n "TableauSite",\n "TableauFlow",\n "LookerDashboard",\n "LookerExplore",\n "LookerLook",\n "LookerView",\n "LookerModel",\n "LookerProject",\n "LookerQuery",\n "LookerFolder",\n "LookerTile",\n "LookerField",\n "SigmaWorkbook",\n "SigmaDataset",\n "SigmaDataElement",\n "SigmaPage",\n "ThoughtspotLiveboard",\n "ThoughtspotAnswer",\n "ThoughtspotWorksheet",\n "MetabaseDashboard",\n "MetabaseQuestion",\n "MetabaseCollection",\n "QuickSightDashboard",\n "QuickSightDataset",\n "QuickSightAnalysis",\n "PresetDashboard",\n "PresetChart",\n "PresetDataset",\n "MicroStrategyDossier",\n "MicroStrategyReport",\n "ModeReport",\n "ModeQuery",\n "ModeWorkspace",\n "DomoDashboard",\n "DomoDataset",\n "DomoCard",\n "SisenseDashboard",\n "SisenseDatamodel",\n "RedashDashboard",\n "RedashQuery",\n "QlikApp",\n "QlikDataset",\n "QlikSheet",\n "QlikSpace"\n ],\n "type": "string"\n },\n {\n "items": {\n "enum": [\n "Table",\n "Column",\n "View",\n "Database",\n "Schema",\n "MaterializedView",\n "AtlasGlossary",\n "AtlasGlossaryTerm",\n "AtlasGlossaryCategory",\n "Connection",\n "Process",\n "Query",\n "Dashboard",\n "Report",\n "DataDomain",\n "DataProduct",\n "PowerBIReport",\n "PowerBIDataset",\n "PowerBIDashboard",\n "PowerBIWorkspace",\n "PowerBIDataflow",\n "PowerBITable",\n "PowerBIMeasure",\n "PowerBIColumn",\n "PowerBIPage",\n "DbtModel",\n "DbtTest",\n "DbtSource",\n "DbtMetric",\n "DbtModelColumn",\n "TableauDashboard",\n "TableauCalculatedField",\n "TableauWorkbook",\n "TableauDatasource",\n "TableauProject",\n "TableauWorksheet",\n "TableauSite",\n "TableauFlow",\n "LookerDashboard",\n "LookerExplore",\n "LookerLook",\n "LookerView",\n "LookerModel",\n "LookerProject",\n "LookerQuery",\n "LookerFolder",\n "LookerTile",\n "LookerField",\n "SigmaWorkbook",\n "SigmaDataset",\n "SigmaDataElement",\n "SigmaPage",\n "ThoughtspotLiveboard",\n "ThoughtspotAnswer",\n "ThoughtspotWorksheet",\n "MetabaseDashboard",\n "MetabaseQuestion",\n "MetabaseCollection",\n "QuickSightDashboard",\n "QuickSightDataset",\n "QuickSightAnalysis",\n "PresetDashboard",\n "PresetChart",\n "PresetDataset",\n "MicroStrategyDossier",\n "MicroStrategyReport",\n "ModeReport",\n "ModeQuery",\n "ModeWorkspace",\n "DomoDashboard",\n "DomoDataset",\n "DomoCard",\n "SisenseDashboard",\n "SisenseDatamodel",\n "RedashDashboard",\n "RedashQuery",\n "QlikApp",\n "QlikDataset",\n "QlikSheet",\n "QlikSpace"\n ],\n "type": "string"\n },\n "type": "array"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Asset type(s) to search. Single type or list of types."\n }' + ), + ] = None, + glossary_qualified_name: Annotated[ + str | None, + cyclopts.Parameter( + help='Glossary qualifiedName to scope search to terms/categories within that glossary\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Glossary qualifiedName to scope search to terms/categories within that glossary"\n }' + ), + ] = None, + include_archived: Annotated[ + bool, cyclopts.Parameter(help="Include archived/deleted assets") + ] = False, + limit: Annotated[ + int, cyclopts.Parameter(help="Max results to return (default 10, max 20)") + ] = 10, + offset: Annotated[int, cyclopts.Parameter(help="Pagination offset")] = 0, + sort_by: Annotated[ + str | None, + cyclopts.Parameter( + help='Single field to sort by\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Single field to sort by"\n }' + ), + ] = None, + sort_order: Annotated[str, cyclopts.Parameter(help="Sort order")] = "ASC", + sort: Annotated[ + str | None, + cyclopts.Parameter( + help='Multi-field sort\\nJSON Schema: {\n "anyOf": [\n {\n "items": {},\n "type": "array"\n },\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Multi-field sort",\n "examples": [\n [\n {\n "field": "name",\n "order": "ASC"\n }\n ],\n [\n {\n "field": "createTime",\n "order": "DESC"\n },\n {\n "field": "name",\n "order": "ASC"\n }\n ]\n ]\n }' + ), + ] = None, + connection_qualified_name: Annotated[ + str | None, + cyclopts.Parameter( + help='Filter by connection (e.g., default/snowflake/123)\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Filter by connection (e.g., default/snowflake/123)"\n }' + ), + ] = None, + tags: Annotated[ + str | None, + cyclopts.Parameter( + help='Filter by Atlan tag names\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "type": "string"\n },\n "type": "array"\n },\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Filter by Atlan tag names"\n }' + ), + ] = None, + directly_tagged: Annotated[ + bool, cyclopts.Parameter(help="Only directly tagged (not inherited)") + ] = True, + domain_guids: Annotated[ + str | None, + cyclopts.Parameter( + help='Filter by domain GUIDs. For DataProduct/DataDomain asset types, this recursively includes all sub-domains — e.g. passing a parent domain GUID returns products under that domain AND all its sub-domains at any depth.\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "type": "string"\n },\n "type": "array"\n },\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Filter by domain GUIDs. For DataProduct/DataDomain asset types, this recursively includes all sub-domains — e.g. passing a parent domain GUID returns products under that domain AND all its sub-domains at any depth."\n }' + ), + ] = None, + date_range: Annotated[ + str | None, + cyclopts.Parameter( + help='Date filters\\nJSON Schema: {\n "anyOf": [\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Date filters",\n "examples": [\n {\n "createTime": {\n "gte": 1704067200000\n }\n },\n {\n "updateTime": {\n "gte": 1704067200000,\n "lte": 1706745600000\n }\n }\n ]\n }' + ), + ] = None, + guids: Annotated[ + str | None, + cyclopts.Parameter( + help='Filter by specific asset GUIDs\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "type": "string"\n },\n "type": "array"\n },\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Filter by specific asset GUIDs"\n }' + ), + ] = None, + term_guids: Annotated[ + str | None, + cyclopts.Parameter( + help='Filter by assigned glossary term GUIDs\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "type": "string"\n },\n "type": "array"\n },\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Filter by assigned glossary term GUIDs"\n }' + ), + ] = None, + aggregations: Annotated[ + str | None, + cyclopts.Parameter( + help='Aggregations to compute. Use __meanings aggregation to find which glossary terms are linked to assets.\\nJSON Schema: {\n "anyOf": [\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Aggregations to compute. Use __meanings aggregation to find which glossary terms are linked to assets.",\n "examples": [\n {\n "by_type": {\n "field": "__typeName",\n "size": 10,\n "type": "terms"\n }\n },\n {\n "by_owner": {\n "field": "ownerUsers",\n "size": 20,\n "type": "terms"\n }\n },\n {\n "linked_terms": {\n "field": "__meanings",\n "size": 500\n }\n }\n ]\n }' + ), + ] = None, + count_only: Annotated[ + bool, cyclopts.Parameter(help="Return only count, no results") + ] = False, + scroll: Annotated[ + bool, cyclopts.Parameter(help="Enable scroll mode for >10k results") + ] = False, + include_readme: Annotated[ + bool, + cyclopts.Parameter( + help="Set true to fetch and return README content for each asset. Only enable when the user explicitly asks for README/documentation content — adds latency." + ), + ] = False, + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """BACKUP search tool — only use when the primary search tools (start_deep_query_tool or semantic_search_tool) cannot express what you need. Provides structured asset search with exact filters, conditions, aggregations, sorting, and pagination. Use this ONLY for precise attribute filters (e.g. certificateStatus, connectorName), term_guids, tags, domain_guids, count_only, or aggregations that the primary tools don\'t support. + + LINKED/UNLINKED TERMS — MANDATORY 2-CALL FLOW: + When the user asks about linked, unlinked, orphan, or unused glossary terms you MUST make TWO separate calls: + Call 1: Get terms — search_assets(asset_type="AtlasGlossaryTerm", glossary_qualified_name="", include_attributes=["qualifiedName","name"], limit=100) + Call 2: Get linked term QNs — search_assets(aggregations={"linked_terms": {"field": "__meanings", "size": 500}}, limit=1) — NO asset_type, NO glossary filter. This aggregates __meanings across ALL assets in the catalog. + Then DIFF: terms from Call 1 whose qualifiedName appears in Call 2\'s aggregation buckets are LINKED. Terms absent are UNLINKED. + IMPORTANT: Call 2 must NOT have asset_type or glossary_qualified_name filters — __meanings lives on regular assets (Tables, Columns), not on terms. Scoping Call 2 to terms returns empty buckets. + + GLOSSARY TERM WORKFLOWS: + - Filter terms by category: conditions={"__categories": ""}. NOTE: __categories only stores the DIRECT parent category. To include subcategories, first fetch all categories in the glossary (asset_type="AtlasGlossaryCategory", include_attributes=["qualifiedName","parentCategory"]), walk the parentCategory tree to find all descendants, then use conditions={"__categories": {"operator": "within", "value": [allDescendantQNs]}}. + NOT for users/groups — use resolve_metadata_tool for those.""" + conditions_parsed = _maybe_json(conditions) + negative_conditions_parsed = _maybe_json(negative_conditions) + some_conditions_parsed = _maybe_json(some_conditions) + include_attributes_parsed = _maybe_json(include_attributes) + asset_type_parsed = _maybe_json(asset_type) + sort_parsed = _maybe_json(sort) + tags_parsed = _maybe_json(tags) + domain_guids_parsed = _maybe_json(domain_guids) + date_range_parsed = _maybe_json(date_range) + guids_parsed = _maybe_json(guids) + term_guids_parsed = _maybe_json(term_guids) + aggregations_parsed = _maybe_json(aggregations) + + await _call_tool( + "search_assets_tool", + { + "conditions": conditions_parsed, + "negative_conditions": negative_conditions_parsed, + "some_conditions": some_conditions_parsed, + "min_somes": min_somes, + "include_attributes": include_attributes_parsed, + "asset_type": asset_type_parsed, + "glossary_qualified_name": glossary_qualified_name, + "include_archived": include_archived, + "limit": limit, + "offset": offset, + "sort_by": sort_by, + "sort_order": sort_order, + "sort": sort_parsed, + "connection_qualified_name": connection_qualified_name, + "tags": tags_parsed, + "directly_tagged": directly_tagged, + "domain_guids": domain_guids_parsed, + "date_range": date_range_parsed, + "guids": guids_parsed, + "term_guids": term_guids_parsed, + "aggregations": aggregations_parsed, + "count_only": count_only, + "scroll": scroll, + "include_readme": include_readme, + "user_query": user_query, + }, + ) + + +@app.command(name="traverse_lineage_tool") +async def traverse_lineage_tool( + *, + guid: Annotated[str, cyclopts.Parameter(help="GUID of the starting asset")], + direction: Annotated[str, cyclopts.Parameter(help="Lineage traversal direction")], + depth: Annotated[ + int, cyclopts.Parameter(help="Maximum depth to traverse") + ] = 1000000, + size: Annotated[ + int, + cyclopts.Parameter( + help="Maximum number of data assets to show in the lineage graph. The widget supports progressive loading — users can click expand on leaf nodes to load more. Keep this small (5-15). Max 20." + ), + ] = 10, + immediate_neighbors: Annotated[ + bool, cyclopts.Parameter(help="Only return immediate neighbors (one hop)") + ] = True, + offset: Annotated[ + int, cyclopts.Parameter(help="Pagination offset for relations (default 0)") + ] = 0, + include_attributes: Annotated[ + str | None, + cyclopts.Parameter( + help='Additional attributes to include\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Additional attributes to include"\n }' + ), + ] = None, + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Traverse upstream or downstream lineage from an asset. + + Returns a widget-ready graph with data assets connected by lineage edges. + Process connector nodes are bridged in the graph (A -> B instead of + A -> Process -> B); their entity data is exposed in the ``process_map`` + field and each relation carries ``via_process_guids`` identifying which + ETL/transformation processes produced the connection. The lineage widget + supports progressive loading — leaf nodes show an expand button that + fetches deeper lineage on demand. Keep size small (5-15) for responsive + results; do NOT request all lineage at once.""" + + await _call_tool( + "traverse_lineage_tool", + { + "guid": guid, + "direction": direction, + "depth": depth, + "size": size, + "immediate_neighbors": immediate_neighbors, + "offset": offset, + "include_attributes": include_attributes, + "user_query": user_query, + }, + ) + + +@app.command(name="query_assets_tool") +async def query_assets_tool( + *, + sql: Annotated[ + str, cyclopts.Parameter(help="SQL query to execute against the connection") + ], + connection_qualified_name: Annotated[ + str, cyclopts.Parameter(help="Qualified name of the connection to query") + ], + default_schema: Annotated[ + str | None, + cyclopts.Parameter( + help='Default schema for unqualified table references\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Default schema for unqualified table references"\n }' + ), + ] = None, + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Execute SQL queries against Atlan connections to preview data.""" + + await _call_tool( + "query_assets_tool", + { + "sql": sql, + "connection_qualified_name": connection_qualified_name, + "default_schema": default_schema, + "user_query": user_query, + }, + ) + + +@app.command(name="resolve_metadata_tool") +async def resolve_metadata_tool( + *, + namespace_type: Annotated[ + str, + cyclopts.Parameter( + help="Metadata namespace to search. Use 'data_domain_and_product' for BOTH data domains AND data products." + ), + ], + query: Annotated[ + str, + cyclopts.Parameter( + help="Search query - name, description, or natural language" + ), + ], + limit: Annotated[ + int, cyclopts.Parameter(help="Max results (default 10, max 20)") + ] = 10, + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Use this to search for users and groups, and to get exact usernames and group names before making updates. Also use for extra discovery on classifications, business_metadata, glossary, and data_domain_and_product when semantic_search doesn\'t return needed results, or before write operations to confirm exact names and GUIDs.""" + + await _call_tool( + "resolve_metadata_tool", + { + "namespace_type": namespace_type, + "query": query, + "limit": limit, + "user_query": user_query, + }, + ) + + +@app.command(name="search_atlan_docs_tool") +async def search_atlan_docs_tool( + *, + query: Annotated[ + str, + cyclopts.Parameter( + help="Question about Atlan features, how-tos, or configuration. E.g. 'How do I connect Snowflake to Atlan?' or 'How do I create a data product?'" + ), + ], + top_k: Annotated[ + int, + cyclopts.Parameter(help="Number of documentation sources to retrieve (1-10)"), + ] = 5, + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Search Atlan\'s customer-facing documentation and return an LLM-generated answer with source citations. Use for how-to questions about Atlan features — not for searching data assets (use semantic_search_tool for that).""" + + await _call_tool( + "search_atlan_docs_tool", + {"query": query, "top_k": top_k, "user_query": user_query}, + ) + + +@app.command(name="get_groups_tool") +async def get_groups_tool( + *, + name_filter: Annotated[ + str | None, + cyclopts.Parameter( + help='Filter by name (partial match)\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Filter by name (partial match)"\n }' + ), + ] = None, + group_id: Annotated[ + str | None, + cyclopts.Parameter( + help='Get specific group by ID (GUID)\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Get specific group by ID (GUID)"\n }' + ), + ] = None, + include_members: Annotated[ + bool, cyclopts.Parameter(help="Include group members in response") + ] = False, + limit: Annotated[int, cyclopts.Parameter(help="Maximum results (1-250)")] = 50, + offset: Annotated[int, cyclopts.Parameter(help="Pagination offset")] = 0, + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Get workspace groups and their members. Use include_members=True to list users in a group.""" + + await _call_tool( + "get_groups_tool", + { + "name_filter": name_filter, + "group_id": group_id, + "include_members": include_members, + "limit": limit, + "offset": offset, + "user_query": user_query, + }, + ) + + +@app.command(name="get_asset_tool") +async def get_asset_tool( + *, + guid: Annotated[ + str | None, + cyclopts.Parameter( + help='Asset GUID to retrieve\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Asset GUID to retrieve"\n }' + ), + ] = None, + qualified_name: Annotated[ + str | None, + cyclopts.Parameter( + help='Asset qualified name to retrieve (e.g. default/snowflake/1234/DB/SCHEMA/TABLE)\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Asset qualified name to retrieve (e.g. default/snowflake/1234/DB/SCHEMA/TABLE)"\n }' + ), + ] = None, + asset_type: Annotated[ + str | None, + cyclopts.Parameter( + help='Asset type (required when using qualified_name). e.g. Table, Column, View, Database, Schema\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Asset type (required when using qualified_name). e.g. Table, Column, View, Database, Schema"\n }' + ), + ] = None, + include_attributes: Annotated[ + str | None, + cyclopts.Parameter( + help='Additional attributes to include\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Additional attributes to include"\n }' + ), + ] = None, + include_dq_checks: Annotated[ + bool, + cyclopts.Parameter( + help="Set true to include linked data quality checks (Soda, Anomalo, Monte Carlo, Atlan native DQ rules) for this asset" + ), + ] = False, + include_readme: Annotated[ + bool, + cyclopts.Parameter( + help="Set true to fetch and return README content for the asset. Only enable when the user explicitly asks for README/documentation content — adds latency." + ), + ] = False, + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Get detailed information about a single asset by its GUID or qualified name.""" + + await _call_tool( + "get_asset_tool", + { + "guid": guid, + "qualified_name": qualified_name, + "asset_type": asset_type, + "include_attributes": include_attributes, + "include_dq_checks": include_dq_checks, + "include_readme": include_readme, + "user_query": user_query, + }, + ) + + +@app.command(name="update_assets_tool") +async def update_assets_tool( + *, + updates: Annotated[ + str, + cyclopts.Parameter( + help='Asset update(s). Each update is a self-contained dict with asset identity (guid, name, qualified_name, type_name) plus the fields to change: user_description, certificate_status, readme, term, owner_users, owner_groups, category_guids, term_relations. Multiple attributes can be updated per asset in one call. For owner_users/owner_groups: REPLACES the full list — include existing owners to keep them. Use exact usernames from resolve_metadata. For category_guids: list of category GUIDs to assign an AtlasGlossaryTerm to (replaces existing categories). For AtlasGlossaryTerm or AtlasGlossaryCategory: also include glossary_guid (the parent glossary GUID, available as glossaryGuid in search results). For term: operation must be exactly \'append\' (link terms), \'remove\' (unlink terms), or \'replace\' (overwrite all). Do NOT use \'add\' or \'delete\' — these are invalid and will fail. For term_relations (AtlasGlossaryTerm only): link one term to another using a relation type. Supported types: synonyms, antonyms, see_also (Related to), preferred_terms (Recommended), preferred_to_terms, translated_terms (Translates to), valid_values, valid_values_for, classifies, is_a (Classified by), replaced_by, replacement_terms. Each relation type takes {op, guids} where op is \'append\', \'replace\', or \'remove\' and guids is a list of target term GUIDs. Use search_assets_tool to find GUIDs of both source and target terms before calling this tool.\\nJSON Schema: {\n "anyOf": [\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "items": {},\n "type": "array"\n },\n {\n "type": "string"\n }\n ],\n "description": "Asset update(s). Each update is a self-contained dict with asset identity (guid, name, qualified_name, type_name) plus the fields to change: user_description, certificate_status, readme, term, owner_users, owner_groups, category_guids, term_relations. Multiple attributes can be updated per asset in one call. For owner_users/owner_groups: REPLACES the full list — include existing owners to keep them. Use exact usernames from resolve_metadata. For category_guids: list of category GUIDs to assign an AtlasGlossaryTerm to (replaces existing categories). For AtlasGlossaryTerm or AtlasGlossaryCategory: also include glossary_guid (the parent glossary GUID, available as glossaryGuid in search results). For term: operation must be exactly \'append\' (link terms), \'remove\' (unlink terms), or \'replace\' (overwrite all). Do NOT use \'add\' or \'delete\' — these are invalid and will fail. For term_relations (AtlasGlossaryTerm only): link one term to another using a relation type. Supported types: synonyms, antonyms, see_also (Related to), preferred_terms (Recommended), preferred_to_terms, translated_terms (Translates to), valid_values, valid_values_for, classifies, is_a (Classified by), replaced_by, replacement_terms. Each relation type takes {op, guids} where op is \'append\', \'replace\', or \'remove\' and guids is a list of target term GUIDs. Use search_assets_tool to find GUIDs of both source and target terms before calling this tool.",\n "examples": [\n {\n "certificate_status": "VERIFIED",\n "guid": "abc-123",\n "name": "customers",\n "qualified_name": "default/snowflake/db/schema/customers",\n "type_name": "Table",\n "user_description": "Customer master data"\n },\n [\n {\n "guid": "abc-123",\n "name": "t1",\n "owner_users": [\n "user1",\n "user2"\n ],\n "qualified_name": "qn1",\n "type_name": "Table"\n },\n {\n "guid": "def-456",\n "name": "t2",\n "qualified_name": "qn2",\n "term": {\n "operation": "append",\n "term_guids": [\n "guid1"\n ]\n },\n "type_name": "Table"\n }\n ],\n [\n {\n "certificate_status": "DRAFT",\n "glossary_guid": "glossary-guid-123",\n "guid": "term-456",\n "name": "Churn Rate",\n "qualified_name": "abc@glossary-qn",\n "type_name": "AtlasGlossaryTerm"\n },\n {\n "glossary_guid": "glossary-guid-123",\n "guid": "term-789",\n "name": "Revenue",\n "qualified_name": "def@glossary-qn",\n "type_name": "AtlasGlossaryTerm",\n "user_description": "Total income"\n }\n ],\n {\n "category_guids": [\n "cat-guid-1",\n "cat-guid-2"\n ],\n "glossary_guid": "glossary-guid-123",\n "guid": "term-101",\n "name": "Net Revenue",\n "qualified_name": "ghi@glossary-qn",\n "type_name": "AtlasGlossaryTerm"\n },\n {\n "glossary_guid": "glossary-guid-123",\n "guid": "term-202",\n "name": "Revenue",\n "qualified_name": "jkl@glossary-qn",\n "term_relations": {\n "antonyms": {\n "guids": [\n "guid-of-cost"\n ],\n "op": "append"\n },\n "synonyms": {\n "guids": [\n "guid-of-income",\n "guid-of-earnings"\n ],\n "op": "append"\n }\n },\n "type_name": "AtlasGlossaryTerm"\n }\n ]\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Update attributes on one or more assets. Each item specifies the asset identity + fields to change. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + updates_parsed = _maybe_json(updates) + + await _call_tool( + "update_assets_tool", + {"updates": updates_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="create_glossaries") +async def create_glossaries( + *, + glossaries: Annotated[ + str, + cyclopts.Parameter( + help='Glossary definitions\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "additionalProperties": true,\n "type": "object"\n },\n "type": "array"\n },\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "type": "string"\n }\n ],\n "description": "Glossary definitions",\n "examples": [\n {\n "name": "Business Glossary",\n "user_description": "Company-wide terms"\n },\n [\n {\n "name": "Glossary A"\n },\n {\n "certificate_status": "VERIFIED",\n "name": "Glossary B"\n }\n ]\n ]\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Create one or more glossaries in Atlan. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + glossaries_parsed = _maybe_json(glossaries) + + await _call_tool( + "create_glossaries", + {"glossaries": glossaries_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="create_glossary_terms") +async def create_glossary_terms( + *, + terms: Annotated[ + str, + cyclopts.Parameter( + help='Term definitions\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "additionalProperties": true,\n "type": "object"\n },\n "type": "array"\n },\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "type": "string"\n }\n ],\n "description": "Term definitions",\n "examples": [\n {\n "glossary_guid": "glossary-guid-123",\n "name": "Revenue",\n "user_description": "Total income"\n },\n [\n {\n "glossary_guid": "g-123",\n "name": "Term A"\n },\n {\n "category_guids": [\n "cat-1"\n ],\n "glossary_guid": "g-123",\n "name": "Term B"\n }\n ]\n ]\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Create one or more glossary terms in Atlan. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + terms_parsed = _maybe_json(terms) + + await _call_tool( + "create_glossary_terms", + {"terms": terms_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="create_glossary_categories") +async def create_glossary_categories( + *, + categories: Annotated[ + str, + cyclopts.Parameter( + help='Category definitions\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "additionalProperties": true,\n "type": "object"\n },\n "type": "array"\n },\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "type": "string"\n }\n ],\n "description": "Category definitions",\n "examples": [\n {\n "glossary_guid": "glossary-guid-123",\n "name": "Finance Terms"\n },\n [\n {\n "glossary_guid": "g-123",\n "name": "Cat A"\n },\n {\n "glossary_guid": "g-123",\n "name": "Cat B",\n "parent_category_guid": "cat-a-guid"\n }\n ]\n ]\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Create one or more glossary categories in Atlan. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + categories_parsed = _maybe_json(categories) + + await _call_tool( + "create_glossary_categories", + {"categories": categories_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="create_domains") +async def create_domains( + *, + domains: Annotated[ + str, + cyclopts.Parameter( + help='Domain definitions\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "additionalProperties": true,\n "type": "object"\n },\n "type": "array"\n },\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "type": "string"\n }\n ],\n "description": "Domain definitions",\n "examples": [\n {\n "name": "Sales Domain",\n "user_description": "Sales analytics data"\n },\n [\n {\n "name": "Finance"\n },\n {\n "name": "Marketing",\n "parent_domain_qualified_name": "default/domain/Finance"\n }\n ]\n ]\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Create one or more data domains in Atlan. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + domains_parsed = _maybe_json(domains) + + await _call_tool( + "create_domains", + {"domains": domains_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="create_data_products") +async def create_data_products( + *, + products: Annotated[ + str, + cyclopts.Parameter( + help='Product definitions\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "additionalProperties": true,\n "type": "object"\n },\n "type": "array"\n },\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "type": "string"\n }\n ],\n "description": "Product definitions",\n "examples": [\n {\n "asset_guids": [\n "asset-guid-1"\n ],\n "domain_qualified_name": "default/domain/Sales",\n "name": "Customer 360"\n },\n [\n {\n "asset_guids": [\n "g1",\n "g2"\n ],\n "domain_qualified_name": "default/domain/D1",\n "name": "Prod A",\n "user_description": "Product A"\n }\n ]\n ]\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Create one or more data products in Atlan. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + products_parsed = _maybe_json(products) + + await _call_tool( + "create_data_products", + {"products": products_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="create_dq_rules_tool") +async def create_dq_rules_tool( + *, + rules: Annotated[ + str, + cyclopts.Parameter( + help='DQ rule definitions. Use EXACTLY these field names: rule_type (str, REQUIRED): Null Count, Null Percentage, Blank Count, Blank Percentage, Min Value, Max Value, Average, Standard Deviation, Unique Count, Duplicate Count, Regex, String Length, Valid Values, Valid Values Reference, Freshness, Row Count, Custom SQL, Recon Row Count, Recon Average, Recon Sum, Recon Duplicate Count, Recon Unique Count. asset_qualified_name (str, REQUIRED): qualified name of the table/view. threshold_value (number, REQUIRED): threshold for the rule. threshold_compare_operator (str): EQUAL (EQ), GREATER_THAN_EQUAL (GTE), LESS_THAN_EQUAL (LTE), BETWEEN (BETWEEN), GREATER_THAN (GT), LESS_THAN (LT). alert_priority (str): URGENT, NORMAL, LOW. column_qualified_name (str): required for column-level rules. custom_sql (str): SQL query, required when rule_type is \'Custom SQL\'. rule_name (str): display name, required when rule_type is \'Custom SQL\'. dimension (str): required when rule_type is \'Custom SQL\': COMPLETENESS, TIMELINESS, ACCURACY, CONSISTENCY, UNIQUENESS, VALIDITY, VOLUME. threshold_unit (str): PERCENTAGE, SECONDS, MINUTES, HOURS, DAYS, WEEKS, MONTHS, YEARS, ABSOLUTE. reference_dataset_qualified_name (str): required for Recon and Reference rules. reference_column_qualified_name (str): required for column-level Recon/Reference rules. description (str): optional rule description.\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "additionalProperties": true,\n "type": "object"\n },\n "type": "array"\n },\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "type": "string"\n }\n ],\n "description": "DQ rule definitions. Use EXACTLY these field names: rule_type (str, REQUIRED): Null Count, Null Percentage, Blank Count, Blank Percentage, Min Value, Max Value, Average, Standard Deviation, Unique Count, Duplicate Count, Regex, String Length, Valid Values, Valid Values Reference, Freshness, Row Count, Custom SQL, Recon Row Count, Recon Average, Recon Sum, Recon Duplicate Count, Recon Unique Count. asset_qualified_name (str, REQUIRED): qualified name of the table/view. threshold_value (number, REQUIRED): threshold for the rule. threshold_compare_operator (str): EQUAL (EQ), GREATER_THAN_EQUAL (GTE), LESS_THAN_EQUAL (LTE), BETWEEN (BETWEEN), GREATER_THAN (GT), LESS_THAN (LT). alert_priority (str): URGENT, NORMAL, LOW. column_qualified_name (str): required for column-level rules. custom_sql (str): SQL query, required when rule_type is \'Custom SQL\'. rule_name (str): display name, required when rule_type is \'Custom SQL\'. dimension (str): required when rule_type is \'Custom SQL\': COMPLETENESS, TIMELINESS, ACCURACY, CONSISTENCY, UNIQUENESS, VALIDITY, VOLUME. threshold_unit (str): PERCENTAGE, SECONDS, MINUTES, HOURS, DAYS, WEEKS, MONTHS, YEARS, ABSOLUTE. reference_dataset_qualified_name (str): required for Recon and Reference rules. reference_column_qualified_name (str): required for column-level Recon/Reference rules. description (str): optional rule description.",\n "examples": [\n {\n "asset_qualified_name": "default/snowflake/db/schema/table",\n "column_qualified_name": "default/snowflake/db/schema/table/column",\n "rule_type": "Null Count",\n "threshold_compare_operator": "EQ",\n "threshold_value": 0\n },\n {\n "asset_qualified_name": "default/snowflake/db/schema/table",\n "rule_type": "Row Count",\n "threshold_compare_operator": "GTE",\n "threshold_value": 100\n },\n {\n "asset_qualified_name": "default/snowflake/db/schema/table",\n "custom_sql": "SELECT COUNT(*) FROM table",\n "dimension": "COMPLETENESS",\n "rule_name": "MyRule",\n "rule_type": "Custom SQL",\n "threshold_compare_operator": "EQ",\n "threshold_value": 0\n },\n {\n "asset_qualified_name": "default/snowflake/db/schema/source_table",\n "reference_dataset_qualified_name": "default/snowflake/db/schema/target_table",\n "rule_type": "Recon Row Count",\n "threshold_unit": "PERCENTAGE",\n "threshold_value": 3\n },\n {\n "asset_qualified_name": "default/snowflake/db/schema/source_table",\n "column_qualified_name": "default/snowflake/db/schema/source_table/col",\n "reference_column_qualified_name": "default/snowflake/db/schema/target_table/col",\n "reference_dataset_qualified_name": "default/snowflake/db/schema/target_table",\n "rule_type": "Recon Unique Count",\n "threshold_unit": "PERCENTAGE",\n "threshold_value": 2\n },\n {\n "asset_qualified_name": "default/snowflake/db/schema/table",\n "column_qualified_name": "default/snowflake/db/schema/table/country",\n "reference_column_qualified_name": "default/snowflake/db/schema/valid_countries/code",\n "reference_dataset_qualified_name": "default/snowflake/db/schema/valid_countries",\n "rule_type": "Valid Values Reference",\n "threshold_value": 2\n }\n ]\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Create data quality rules in Atlan. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + rules_parsed = _maybe_json(rules) + + await _call_tool( + "create_dq_rules_tool", + {"rules": rules_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="schedule_dq_rules_tool") +async def schedule_dq_rules_tool( + *, + schedules: Annotated[ + str, + cyclopts.Parameter( + help='Schedule definitions\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "additionalProperties": true,\n "type": "object"\n },\n "type": "array"\n },\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "type": "string"\n }\n ],\n "description": "Schedule definitions",\n "examples": [\n {\n "asset_name": "customers",\n "asset_qualified_name": "default/snowflake/db/schema/customers",\n "asset_type": "Table",\n "schedule_crontab": "0 6 * * *",\n "schedule_time_zone": "America/New_York"\n },\n [\n {\n "asset_name": "v1",\n "asset_qualified_name": "qn1",\n "asset_type": "View",\n "schedule_crontab": "0 */4 * * *",\n "schedule_time_zone": "UTC"\n }\n ]\n ]\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Schedule data quality rule execution on assets. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + schedules_parsed = _maybe_json(schedules) + + await _call_tool( + "schedule_dq_rules_tool", + {"schedules": schedules_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="update_dq_rules_tool") +async def update_dq_rules_tool( + *, + rules: Annotated[ + str, + cyclopts.Parameter( + help='DQ rule updates. Use EXACTLY these field names: qualified_name (str, REQUIRED): qualified name of the rule to update. rule_type (str, REQUIRED): Null Count, Null Percentage, Blank Count, Blank Percentage, Min Value, Max Value, Average, Standard Deviation, Unique Count, Duplicate Count, Regex, String Length, Valid Values, Valid Values Reference, Freshness, Row Count, Custom SQL, Recon Row Count, Recon Average, Recon Sum, Recon Duplicate Count, Recon Unique Count. asset_qualified_name (str, REQUIRED): qualified name of the table/view. threshold_value (number): new threshold value. threshold_compare_operator (str): EQUAL (EQ), GREATER_THAN_EQUAL (GTE), LESS_THAN_EQUAL (LTE), BETWEEN (BETWEEN), GREATER_THAN (GT), LESS_THAN (LT). threshold_unit (str): PERCENTAGE, SECONDS, MINUTES, HOURS, DAYS, WEEKS, MONTHS, YEARS, ABSOLUTE. alert_priority (str): URGENT, NORMAL, LOW. custom_sql (str): updated SQL query for Custom SQL rules. rule_name (str): updated display name for Custom SQL rules. dimension (str): for Custom SQL rules: COMPLETENESS, TIMELINESS, ACCURACY, CONSISTENCY, UNIQUENESS, VALIDITY, VOLUME. description (str): optional rule description.\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "additionalProperties": true,\n "type": "object"\n },\n "type": "array"\n },\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "type": "string"\n }\n ],\n "description": "DQ rule updates. Use EXACTLY these field names: qualified_name (str, REQUIRED): qualified name of the rule to update. rule_type (str, REQUIRED): Null Count, Null Percentage, Blank Count, Blank Percentage, Min Value, Max Value, Average, Standard Deviation, Unique Count, Duplicate Count, Regex, String Length, Valid Values, Valid Values Reference, Freshness, Row Count, Custom SQL, Recon Row Count, Recon Average, Recon Sum, Recon Duplicate Count, Recon Unique Count. asset_qualified_name (str, REQUIRED): qualified name of the table/view. threshold_value (number): new threshold value. threshold_compare_operator (str): EQUAL (EQ), GREATER_THAN_EQUAL (GTE), LESS_THAN_EQUAL (LTE), BETWEEN (BETWEEN), GREATER_THAN (GT), LESS_THAN (LT). threshold_unit (str): PERCENTAGE, SECONDS, MINUTES, HOURS, DAYS, WEEKS, MONTHS, YEARS, ABSOLUTE. alert_priority (str): URGENT, NORMAL, LOW. custom_sql (str): updated SQL query for Custom SQL rules. rule_name (str): updated display name for Custom SQL rules. dimension (str): for Custom SQL rules: COMPLETENESS, TIMELINESS, ACCURACY, CONSISTENCY, UNIQUENESS, VALIDITY, VOLUME. description (str): optional rule description.",\n "examples": [\n {\n "alert_priority": "URGENT",\n "asset_qualified_name": "default/snowflake/db/schema/table",\n "qualified_name": "default/dq/rule-qn",\n "rule_type": "Null Count",\n "threshold_value": 5\n },\n {\n "asset_qualified_name": "default/snowflake/db/schema/table",\n "qualified_name": "default/dq/rule-qn",\n "rule_type": "Freshness",\n "threshold_unit": "HOURS",\n "threshold_value": 24\n }\n ]\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Update existing data quality rules. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + rules_parsed = _maybe_json(rules) + + await _call_tool( + "update_dq_rules_tool", + {"rules": rules_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="delete_dq_rules_tool") +async def delete_dq_rules_tool( + *, + rule_guids: Annotated[ + str, + cyclopts.Parameter( + help='GUIDs of DQ rules to delete\\nJSON Schema: {\n "anyOf": [\n {\n "items": {\n "type": "string"\n },\n "type": "array"\n },\n {\n "type": "string"\n }\n ],\n "description": "GUIDs of DQ rules to delete"\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Delete data quality rules by their GUIDs. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + rule_guids_parsed = _maybe_json(rule_guids) + + await _call_tool( + "delete_dq_rules_tool", + {"rule_guids": rule_guids_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="manage_asset_lifecycle_tool") +async def manage_asset_lifecycle_tool( + *, + operation: Annotated[ + str, cyclopts.Parameter(help="Lifecycle operation to perform") + ], + guids: Annotated[ + str | None, + cyclopts.Parameter( + help='Asset GUIDs (required for ARCHIVE/PURGE)\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Asset GUIDs (required for ARCHIVE/PURGE)"\n }' + ), + ] = None, + asset_type: Annotated[ + str | None, + cyclopts.Parameter( + help='Asset type (required for RESTORE)\\nJSON Schema: {\n "anyOf": [\n {\n "enum": [\n "Table",\n "Column",\n "View",\n "Database",\n "Schema",\n "MaterializedView",\n "AtlasGlossary",\n "AtlasGlossaryTerm",\n "AtlasGlossaryCategory",\n "Connection",\n "Process",\n "Query",\n "Dashboard",\n "Report",\n "DataDomain",\n "DataProduct",\n "PowerBIReport",\n "PowerBIDataset",\n "PowerBIDashboard",\n "PowerBIWorkspace",\n "PowerBIDataflow",\n "PowerBITable",\n "PowerBIMeasure",\n "PowerBIColumn",\n "PowerBIPage",\n "DbtModel",\n "DbtTest",\n "DbtSource",\n "DbtMetric",\n "DbtModelColumn",\n "TableauDashboard",\n "TableauCalculatedField",\n "TableauWorkbook",\n "TableauDatasource",\n "TableauProject",\n "TableauWorksheet",\n "TableauSite",\n "TableauFlow",\n "LookerDashboard",\n "LookerExplore",\n "LookerLook",\n "LookerView",\n "LookerModel",\n "LookerProject",\n "LookerQuery",\n "LookerFolder",\n "LookerTile",\n "LookerField",\n "SigmaWorkbook",\n "SigmaDataset",\n "SigmaDataElement",\n "SigmaPage",\n "ThoughtspotLiveboard",\n "ThoughtspotAnswer",\n "ThoughtspotWorksheet",\n "MetabaseDashboard",\n "MetabaseQuestion",\n "MetabaseCollection",\n "QuickSightDashboard",\n "QuickSightDataset",\n "QuickSightAnalysis",\n "PresetDashboard",\n "PresetChart",\n "PresetDataset",\n "MicroStrategyDossier",\n "MicroStrategyReport",\n "ModeReport",\n "ModeQuery",\n "ModeWorkspace",\n "DomoDashboard",\n "DomoDataset",\n "DomoCard",\n "SisenseDashboard",\n "SisenseDatamodel",\n "RedashDashboard",\n "RedashQuery",\n "QlikApp",\n "QlikDataset",\n "QlikSheet",\n "QlikSpace"\n ],\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Asset type (required for RESTORE)"\n }' + ), + ] = None, + qualified_name: Annotated[ + str | None, + cyclopts.Parameter( + help='Qualified name (required for RESTORE)\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Qualified name (required for RESTORE)"\n }' + ), + ] = None, + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Manage asset lifecycle: archive, restore, or permanently purge assets. WARNING: PURGE cannot be undone. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + + await _call_tool( + "manage_asset_lifecycle_tool", + { + "operation": operation, + "guids": guids, + "asset_type": asset_type, + "qualified_name": qualified_name, + "mode": mode, + "user_query": user_query, + }, + ) + + +@app.command(name="manage_announcements_tool") +async def manage_announcements_tool( + *, + asset_guids: Annotated[ + str, + cyclopts.Parameter( + help="Asset GUIDs to update (comma-separated or JSON array)" + ), + ], + operation: Annotated[str, cyclopts.Parameter(help="Announcement operation")], + announcement_type: Annotated[ + str | None, + cyclopts.Parameter( + help='Announcement type (required for SET)\\nJSON Schema: {\n "anyOf": [\n {\n "enum": [\n "information",\n "warning",\n "issue"\n ],\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Announcement type (required for SET)"\n }' + ), + ] = None, + title: Annotated[ + str | None, + cyclopts.Parameter( + help='Announcement title (required for SET)\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Announcement title (required for SET)"\n }' + ), + ] = None, + message: Annotated[ + str | None, + cyclopts.Parameter( + help='Announcement message (optional)\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "Announcement message (optional)"\n }' + ), + ] = None, + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Add or remove announcements (information, warning, issue) on assets. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + + await _call_tool( + "manage_announcements_tool", + { + "asset_guids": asset_guids, + "operation": operation, + "announcement_type": announcement_type, + "title": title, + "message": message, + "mode": mode, + "user_query": user_query, + }, + ) + + +@app.command(name="update_custom_metadata_tool") +async def update_custom_metadata_tool( + *, + updates: Annotated[ + str, + cyclopts.Parameter( + help='Single update (dict) or batch updates (list of dicts). Each item requires: guid, custom_metadata_name, attributes. Single: {"guid": "...", "custom_metadata_name": "CM", "attributes": {"k": "v"}}. Batch: [{"guid": "g1", ...}, {"guid": "g2", ...}]\\nJSON Schema: {\n "anyOf": [\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "items": {},\n "type": "array"\n },\n {\n "type": "string"\n }\n ],\n "description": "Single update (dict) or batch updates (list of dicts). Each item requires: guid, custom_metadata_name, attributes. Single: {\\"guid\\": \\"...\\", \\"custom_metadata_name\\": \\"CM\\", \\"attributes\\": {\\"k\\": \\"v\\"}}. Batch: [{\\"guid\\": \\"g1\\", ...}, {\\"guid\\": \\"g2\\", ...}]"\n }' + ), + ], + replace: Annotated[ + bool, + cyclopts.Parameter( + help="If False (default): partial update — only specified attributes are changed. If True: full replacement — ALL attributes in the CM set are replaced; unspecified attributes are cleared. replace=True only supported for single asset (dict input), not batch." + ), + ] = False, + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Update custom metadata on one or more assets. Set replace=True for full replacement, False (default) for partial update. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + updates_parsed = _maybe_json(updates) + + await _call_tool( + "update_custom_metadata_tool", + { + "updates": updates_parsed, + "replace": replace, + "mode": mode, + "user_query": user_query, + }, + ) + + +@app.command(name="remove_custom_metadata_tool") +async def remove_custom_metadata_tool( + *, + guid: Annotated[str, cyclopts.Parameter(help="GUID of the asset")], + custom_metadata_name: Annotated[ + str, cyclopts.Parameter(help="Name of the custom metadata set to remove") + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Remove a custom metadata set from an asset. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + + await _call_tool( + "remove_custom_metadata_tool", + { + "guid": guid, + "custom_metadata_name": custom_metadata_name, + "mode": mode, + "user_query": user_query, + }, + ) + + +@app.command(name="create_custom_metadata_set_tool") +async def create_custom_metadata_set_tool( + *, + sets: Annotated[ + str, + cyclopts.Parameter( + help='Single CM set (dict) or multiple CM sets (list of dicts). Each requires: display_name (str), attributes (list). Optional: description (str). Each attribute requires: display_name, attribute_type (string|int|float|boolean|date|enum|users|groups|url|SQL|long). Optional per attribute: multi_valued (bool), description (str). For enum attributes: provide enum_values (list of allowed strings) to create a new enum, OR options_name (str) to reference an existing Atlan enum. Single: {"display_name": "Data Quality", "attributes": [{"display_name": "Score", "attribute_type": "int"}, {"display_name": "Dimension", "attribute_type": "enum", "enum_values": ["Accuracy", "Completeness"]}]}. Batch: [{"display_name": "CM1", "attributes": [...]}, {"display_name": "CM2", "attributes": [...]}]\\nJSON Schema: {\n "anyOf": [\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "items": {},\n "type": "array"\n },\n {\n "type": "string"\n }\n ],\n "description": "Single CM set (dict) or multiple CM sets (list of dicts). Each requires: display_name (str), attributes (list). Optional: description (str). Each attribute requires: display_name, attribute_type (string|int|float|boolean|date|enum|users|groups|url|SQL|long). Optional per attribute: multi_valued (bool), description (str). For enum attributes: provide enum_values (list of allowed strings) to create a new enum, OR options_name (str) to reference an existing Atlan enum. Single: {\\"display_name\\": \\"Data Quality\\", \\"attributes\\": [{\\"display_name\\": \\"Score\\", \\"attribute_type\\": \\"int\\"}, {\\"display_name\\": \\"Dimension\\", \\"attribute_type\\": \\"enum\\", \\"enum_values\\": [\\"Accuracy\\", \\"Completeness\\"]}]}. Batch: [{\\"display_name\\": \\"CM1\\", \\"attributes\\": [...]}, {\\"display_name\\": \\"CM2\\", \\"attributes\\": [...]}]"\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Create one or more custom metadata sets with typed attributes. Defines the schema; use update_custom_metadata to set values on assets. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + sets_parsed = _maybe_json(sets) + + await _call_tool( + "create_custom_metadata_set_tool", + {"sets": sets_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="delete_custom_metadata_set_tool") +async def delete_custom_metadata_set_tool( + *, + sets: Annotated[ + str, + cyclopts.Parameter( + help='Single CM set display name (string) or list of display names for batch deletion. WARNING: Irreversible — deletes the schema and all stored attribute values on assets. Single: "Data Quality". Batch: ["Data Quality", "Governance"]\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "items": {},\n "type": "array"\n }\n ],\n "description": "Single CM set display name (string) or list of display names for batch deletion. WARNING: Irreversible — deletes the schema and all stored attribute values on assets. Single: \\"Data Quality\\". Batch: [\\"Data Quality\\", \\"Governance\\"]"\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Permanently delete one or more custom metadata sets and all their attribute values from assets. Irreversible. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + sets_parsed = _maybe_json(sets) + + await _call_tool( + "delete_custom_metadata_set_tool", + {"sets": sets_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="add_attributes_to_cm_set_tool") +async def add_attributes_to_cm_set_tool( + *, + display_name: Annotated[ + str, cyclopts.Parameter(help="Display name of the existing CM set") + ], + attributes: Annotated[ + str, + cyclopts.Parameter( + help='Attributes to add. Each requires: display_name, attribute_type (string|int|float|boolean|date|enum|users|groups|url|SQL|long). Optional: multi_valued (bool), description (str). For enum attributes: provide enum_values (list of allowed strings) to create a new enum, OR options_name (str) to reference an existing Atlan enum. Example: [{"display_name": "Risk Score", "attribute_type": "int"}, {"display_name": "Status", "attribute_type": "enum", "enum_values": ["Draft", "Approved", "Rejected"]}]\\nJSON Schema: {\n "anyOf": [\n {\n "items": {},\n "type": "array"\n },\n {\n "type": "string"\n }\n ],\n "description": "Attributes to add. Each requires: display_name, attribute_type (string|int|float|boolean|date|enum|users|groups|url|SQL|long). Optional: multi_valued (bool), description (str). For enum attributes: provide enum_values (list of allowed strings) to create a new enum, OR options_name (str) to reference an existing Atlan enum. Example: [{\\"display_name\\": \\"Risk Score\\", \\"attribute_type\\": \\"int\\"}, {\\"display_name\\": \\"Status\\", \\"attribute_type\\": \\"enum\\", \\"enum_values\\": [\\"Draft\\", \\"Approved\\", \\"Rejected\\"]}]"\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Add new typed attributes to an existing custom metadata set. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + attributes_parsed = _maybe_json(attributes) + + await _call_tool( + "add_attributes_to_cm_set_tool", + { + "display_name": display_name, + "attributes": attributes_parsed, + "mode": mode, + "user_query": user_query, + }, + ) + + +@app.command(name="remove_attributes_from_cm_set_tool") +async def remove_attributes_from_cm_set_tool( + *, + display_name: Annotated[ + str, cyclopts.Parameter(help="Display name of the existing CM set") + ], + attribute_names: Annotated[ + str, + cyclopts.Parameter( + help='Display name(s) of attributes to remove (archive). Archived attributes are soft-deleted: values are cleared from assets and hidden in UI. Example: ["Score", "Reviewed By"] or a JSON string of the list.\\nJSON Schema: {\n "anyOf": [\n {\n "items": {},\n "type": "array"\n },\n {\n "type": "string"\n }\n ],\n "description": "Display name(s) of attributes to remove (archive). Archived attributes are soft-deleted: values are cleared from assets and hidden in UI. Example: [\\"Score\\", \\"Reviewed By\\"] or a JSON string of the list."\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Remove (archive) attributes from an existing custom metadata set. Archived attributes are soft-deleted and their values cleared from all assets. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + attribute_names_parsed = _maybe_json(attribute_names) + + await _call_tool( + "remove_attributes_from_cm_set_tool", + { + "display_name": display_name, + "attribute_names": attribute_names_parsed, + "mode": mode, + "user_query": user_query, + }, + ) + + +@app.command(name="add_atlan_tags_tool") +async def add_atlan_tags_tool( + *, + updates: Annotated[ + str, + cyclopts.Parameter( + help='Single tag addition (dict) or batch (list of dicts). Each item: {guid, tag_names, propagate (default true), remove_on_delete (default true), restrict_lineage_propagation (default false), restrict_hierarchy_propagation (default false)}. Single: {"guid": "...", "tag_names": ["PII"]}. Batch: [{"guid": "g1", "tag_names": ["PII"]}, ...]\\nJSON Schema: {\n "anyOf": [\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "items": {},\n "type": "array"\n },\n {\n "type": "string"\n }\n ],\n "description": "Single tag addition (dict) or batch (list of dicts). Each item: {guid, tag_names, propagate (default true), remove_on_delete (default true), restrict_lineage_propagation (default false), restrict_hierarchy_propagation (default false)}. Single: {\\"guid\\": \\"...\\", \\"tag_names\\": [\\"PII\\"]}. Batch: [{\\"guid\\": \\"g1\\", \\"tag_names\\": [\\"PII\\"]}, ...]"\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Add Atlan tags to one or more assets. Pass a dict for single asset, list for batch. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + updates_parsed = _maybe_json(updates) + + await _call_tool( + "add_atlan_tags_tool", + {"updates": updates_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="remove_atlan_tag_tool") +async def remove_atlan_tag_tool( + *, + updates: Annotated[ + str, + cyclopts.Parameter( + help='Single tag removal (dict) or batch (list of dicts). Each item: {guid, tag_name}. Single: {"guid": "...", "tag_name": "PII"}. Batch: [{"guid": "g1", "tag_name": "PII"}, ...]\\nJSON Schema: {\n "anyOf": [\n {\n "additionalProperties": true,\n "type": "object"\n },\n {\n "items": {},\n "type": "array"\n },\n {\n "type": "string"\n }\n ],\n "description": "Single tag removal (dict) or batch (list of dicts). Each item: {guid, tag_name}. Single: {\\"guid\\": \\"...\\", \\"tag_name\\": \\"PII\\"}. Batch: [{\\"guid\\": \\"g1\\", \\"tag_name\\": \\"PII\\"}, ...]",\n "examples": [\n {\n "guid": "asset-guid-1",\n "tag_name": "PII"\n },\n [\n {\n "guid": "g1",\n "tag_name": "Sensitive"\n },\n {\n "guid": "g2",\n "tag_name": "Confidential"\n }\n ]\n ]\n }' + ), + ], + mode: Annotated[ + str, + cyclopts.Parameter( + help="ALWAYS use 'propose'. Returns a preview for user review. NEVER use 'execute' unless the user has explicitly approved." + ), + ] = "propose", + user_query: Annotated[ + str | None, + cyclopts.Parameter( + help='REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability.\\nJSON Schema: {\n "anyOf": [\n {\n "type": "string"\n },\n {\n "type": "null"\n }\n ],\n "default": null,\n "description": "REQUIRED: Always pass the user\'s exact question/prompt that triggered this tool call. Used for tracing and observability."\n }' + ), + ] = None, +) -> None: + """Remove an Atlan tag from one or more assets. Pass a dict for single asset, list for batch. Always uses propose mode — STOP after proposing and wait for user approval before executing.""" + updates_parsed = _maybe_json(updates) + + await _call_tool( + "remove_atlan_tag_tool", + {"updates": updates_parsed, "mode": mode, "user_query": user_query}, + ) + + +@app.command(name="get_asset_icons") +async def get_asset_icons( + *, + icon_names: Annotated[list[str], cyclopts.Parameter(help="")], +) -> None: + """Fetch SVG icons for asset types. Internal tool used by widgets.""" + await _call_tool("get_asset_icons", {"icon_names": icon_names}) + + +def main() -> None: + global _JSON_MODE + + args = sys.argv[1:] + subcommand = next((a for a in args if not a.startswith("-")), None) + is_login = subcommand == "login" + + new_args: list[str] = [] + i = 0 + while i < len(args): + a = args[i] + if a == "--oauth": + os.environ["_ATLAN_OVERRIDE_OAUTH"] = "1" + # Don't add to new_args — consume the flag globally + elif a == "--json": + _JSON_MODE = True + # Don't add to new_args — consume globally + elif a == "--api-key" and i + 1 < len(args) and not is_login: + os.environ["_ATLAN_OVERRIDE_API_KEY"] = args[i + 1] + i += 1 # skip value too + elif a == "--tenant" and i + 1 < len(args) and not is_login: + os.environ["_ATLAN_OVERRIDE_TENANT"] = args[i + 1] + i += 1 # skip value too + else: + new_args.append(a) + i += 1 + + sys.argv = [sys.argv[0]] + new_args + try: + app() + except SystemExit: + raise + except Exception as exc: + _err(f"[bold red]Error:[/bold red] {exc}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/mcp-cli/pyproject.toml b/mcp-cli/pyproject.toml new file mode 100644 index 0000000..1c6ab46 --- /dev/null +++ b/mcp-cli/pyproject.toml @@ -0,0 +1,48 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "atlan-cli" +dynamic = ["version"] +description = "CLI for calling Atlan MCP tools directly from your terminal" +readme = "README.md" +requires-python = ">=3.10" +license = { text = "Apache-2.0" } +keywords = ["atlan", "mcp", "cli", "data-catalog"] +classifiers = [ + "Development Status :: 3 - Alpha", + "Environment :: Console", + "Intended Audience :: Developers", + "License :: OSI Approved :: Apache Software License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", +] +dependencies = [ + "cyclopts>=3.0", + "fastmcp>=2.14,<3", + "python-dotenv>=1.0", + "keyring>=25.0", + "rich>=13.0", + "httpx>=0.27", +] + +[project.urls] +Homepage = "https://github.com/atlanhq/agent-toolkit" +Repository = "https://github.com/atlanhq/agent-toolkit" +"Bug Tracker" = "https://github.com/atlanhq/agent-toolkit/issues" + +[project.scripts] +atlan = "atlan_cli:main" + +[tool.hatch.version] +path = "atlan_cli.py" + +[tool.hatch.build.targets.wheel] +include = ["atlan_cli.py"] + +[tool.hatch.build.targets.sdist] +include = ["atlan_cli.py", "README.md", "pyproject.toml"]