added audit_history tool

Author: rahul-madaan

modelcontextprotocol/server.py

@@ -14,6 +14,7 @@
     UpdatableAttribute,
     CertificateStatus,
     UpdatableAsset,
+    get_asset_history,
 )
 from pyatlan.model.lineage import LineageDirection
 from utils.parameters import (
@@ -508,6 +509,65 @@ def update_assets_tool(
         }
 
 
+@mcp.tool()
+def get_asset_history_tool(
+    guid=None,
+    qualified_name=None,
+    type_name=None,
+    size=10,
+    sort_order="DESC",
+    include_attributes=None,
+):
+    """
+    Get the audit history of an asset by GUID or qualified name.
+
+    Args:
+        guid (str, optional): GUID of the asset to get history for.
+            Either guid or qualified_name must be provided.
+        qualified_name (str, optional): Qualified name of the asset to get history for.
+            Either guid or qualified_name must be provided.
+        type_name (str, optional): Type name of the asset (required when using qualified_name).
+            Examples: "Table", "Column", "DbtModel", "AtlasGlossary"
+        size (int): Number of history entries to return. Defaults to 10.
+        sort_order (str): Sort order for results. "ASC" for oldest first, "DESC" for newest first.
+            Defaults to "DESC".
+        include_attributes (List[Union[str, AtlanField]], optional): List of additional attributes to include in results.
+            Can be string attribute names or AtlanField objects. These will be added to the default set.
+
+    Returns:
+        Dict[str, Any]: Dictionary containing:
+            - entityAudits: List of audit entries with details about each change
+            - count: Number of audit entries returned
+            - totalCount: Total number of audit entries available
+            - errors: List of any errors encountered
+
+    Examples:
+        # Get history by GUID
+        history = get_asset_history_tool(
+            guid="6fc01478-1263-42ae-b8ca-c4a57da51392",
+            size=20,
+            sort_order="DESC",
+            include_attributes=["owner_users", "description"]
+        )
+
+        # Get history by qualified name
+        history = get_asset_history_tool(
+            qualified_name="default/dbt/1755018137/account/258239/project/376530/model.simple_column_lineage.order_maths",
+            type_name="DbtModel",
+            size=15,
+            sort_order="ASC"
+        )
+    """
+    return get_asset_history(
+        guid=guid,
+        qualified_name=qualified_name,
+        type_name=type_name,
+        size=size,
+        sort_order=sort_order,
+        include_attributes=include_attributes,
+    )
+
+
 @mcp.tool()
 def create_glossaries(glossaries) -> List[Dict[str, Any]]:
     """

modelcontextprotocol/tools/__init__.py

@@ -1,7 +1,7 @@
 from .search import search_assets
 from .dsl import get_assets_by_dsl
 from .lineage import traverse_lineage
-from .assets import update_assets
+from .assets import update_assets, get_asset_history
 from .glossary import (
     create_glossary_category_assets,
     create_glossary_assets,
@@ -21,6 +21,7 @@
     "get_assets_by_dsl",
     "traverse_lineage",
     "update_assets",
+    "get_asset_history",
     "create_glossary_category_assets",
     "create_glossary_assets",
     "create_glossary_term_assets",

modelcontextprotocol/tools/assets.py

@@ -1,9 +1,17 @@
 import logging
-from typing import List, Union, Dict, Any
+from typing import List, Union, Dict, Any, Optional
 from client import get_atlan_client
 from .models import UpdatableAsset, UpdatableAttribute, CertificateStatus
 from pyatlan.model.assets import Readme
 from pyatlan.model.fluent_search import CompoundQuery, FluentSearch
+from pyatlan.model.fields.atlan_fields import AtlanField
+from utils.asset_history import (
+    validate_asset_history_params,
+    create_audit_search_request,
+    process_audit_result,
+    create_sort_item,
+    convert_attributes_to_camel_case,
+)
 
 # Initialize logging
 logger = logging.getLogger(__name__)
@@ -128,3 +136,96 @@ def update_assets(
         error_msg = f"Error updating assets: {str(e)}"
         logger.error(error_msg)
         return {"updated_count": 0, "errors": [error_msg]}
+
+
+def get_asset_history(
+    guid: Optional[str] = None,
+    qualified_name: Optional[str] = None,
+    type_name: Optional[str] = None,
+    size: int = 10,
+    sort_order: str = "DESC",
+    include_attributes: Optional[List[Union[str, AtlanField]]] = None,
+) -> Dict[str, Any]:
+    """
+    Get the audit history of an asset by GUID or qualified name.
+
+    Args:
+        guid (Optional[str]): GUID of the asset to get history for.
+            Either guid or qualified_name must be provided.
+        qualified_name (Optional[str]): Qualified name of the asset to get history for.
+            Either guid or qualified_name must be provided.
+        type_name (Optional[str]): Type name of the asset (required when using qualified_name).
+            Examples: "Table", "Column", "DbtModel", "AtlasGlossary"
+        size (int): Number of history entries to return. Defaults to 10.
+        sort_order (str): Sort order for results. "ASC" for oldest first, "DESC" for newest first.
+            Defaults to "DESC".
+        include_attributes (List[Union[str, AtlanField]], optional): List of additional attributes to include in results.
+            Can be string attribute names or AtlanField objects. These will be added to the default set.
+
+    Returns:
+        Dict[str, Any]: Dictionary containing:
+            - entityAudits: List of audit entries
+            - count: Number of audit entries returned
+            - totalCount: Total number of audit entries available
+            - errors: List of any errors encountered
+
+    Raises:
+        Exception: If there's an error retrieving the asset history
+    """
+    try:
+        # Validate input parameters
+        validation_error = validate_asset_history_params(
+            guid, qualified_name, type_name, sort_order
+        )
+        if validation_error:
+            logger.error(validation_error)
+            return {
+                "errors": [validation_error],
+                "entityAudits": [],
+                "count": 0,
+                "totalCount": 0,
+            }
+
+        logger.info(
+            f"Retrieving asset history with parameters: guid={guid}, qualified_name={qualified_name}, size={size}"
+        )
+
+        # Get Atlan client
+        client = get_atlan_client()
+
+        # Create sort item
+        sort_item = create_sort_item(sort_order)
+
+        # Convert include_attributes from snake_case to camelCase if needed
+        if include_attributes:
+            include_attributes = convert_attributes_to_camel_case(include_attributes)
+
+        # Create and execute audit search request
+        request = create_audit_search_request(
+            guid, qualified_name, type_name, size, sort_item, include_attributes
+        )
+        response = client.audit.search(criteria=request, bulk=False)
+
+        # Process audit results - use current_page() to respect size parameter
+        entity_audits = [
+            process_audit_result(result, include_attributes)
+            for result in response.current_page()
+        ]
+
+        result_data = {
+            "entityAudits": entity_audits,
+            "count": len(entity_audits),
+            "totalCount": response.total_count,
+            "errors": [],
+        }
+
+        logger.info(
+            f"Successfully retrieved {len(entity_audits)} audit entries for asset"
+        )
+        return result_data
+
+    except Exception as e:
+        error_msg = f"Error retrieving asset history: {str(e)}"
+        logger.error(error_msg)
+        logger.exception("Exception details:")
+        return {"errors": [error_msg], "entityAudits": [], "count": 0, "totalCount": 0}