updated, addressed comments

Author: rahul-madaan

modelcontextprotocol/server.py

@@ -511,12 +511,7 @@ 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,
+    guid=None, qualified_name=None, type_name=None, size=10, sort_order="DESC"
 ):
     """
     Get the audit history of an asset by GUID or qualified name.
@@ -531,8 +526,6 @@ def get_asset_history_tool(
         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:
@@ -547,7 +540,6 @@ def 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
@@ -564,7 +556,6 @@ def get_asset_history_tool(
         type_name=type_name,
         size=size,
         sort_order=sort_order,
-        include_attributes=include_attributes,
     )
 
 

modelcontextprotocol/tools/assets.py

@@ -4,13 +4,11 @@
 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
@@ -144,7 +142,6 @@ def get_asset_history(
     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.
@@ -159,8 +156,6 @@ def get_asset_history(
         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:
@@ -196,20 +191,15 @@ def get_asset_history(
         # 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
+            guid, qualified_name, type_name, size, sort_item
         )
         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()
+            process_audit_result(result) for result in response.current_page()
         ]
 
         result_data = {

modelcontextprotocol/utils/asset_history.py

@@ -5,34 +5,15 @@
 """
 
 import logging
-from typing import List, Union, Dict, Any, Optional
+from typing import Dict, Any, Optional
 from pyatlan.client.audit import AuditSearchRequest
 from pyatlan.model.search import DSL, Bool, SortItem, Term
 from pyatlan.model.enums import SortOrder
-from pyatlan.model.fields.atlan_fields import AtlanField
 
 # Initialize logging
 logger = logging.getLogger(__name__)
 
 
-def snake_to_camel(s: str) -> str:
-    """
-    Convert a snake_case string to camelCase.
-
-    Args:
-        s (str): The snake_case string.
-
-    Returns:
-        str: The camelCase string.
-    """
-    parts = s.split("_")
-    return (
-        parts[0] + "".join(word.capitalize() for word in parts[1:])
-        if len(parts) > 1
-        else s
-    )
-
-
 def validate_asset_history_params(
     guid: Optional[str],
     qualified_name: Optional[str],
@@ -69,7 +50,6 @@ def create_audit_search_request(
     type_name: Optional[str],
     size: int,
     sort_item: SortItem,
-    include_attributes: Optional[List[Union[str, AtlanField]]],
 ) -> AuditSearchRequest:
     """
     Create an AuditSearchRequest based on the provided parameters.
@@ -80,7 +60,6 @@ def create_audit_search_request(
         type_name: Asset type name
         size: Number of results to return
         sort_item: Sort configuration
-        include_attributes: Attributes to include in results
 
     Returns:
         Configured AuditSearchRequest
@@ -107,11 +86,7 @@ def create_audit_search_request(
             f"Created audit search request by qualified name: {qualified_name}"
         )
 
-    # Only pass attributes if they are provided
-    if include_attributes:
-        return AuditSearchRequest(dsl=dsl, attributes=include_attributes)
-    else:
-        return AuditSearchRequest(dsl=dsl)
+    return AuditSearchRequest(dsl=dsl)
 
 
 def extract_basic_audit_info(result) -> Dict[str, Any]:
@@ -135,29 +110,21 @@ def extract_basic_audit_info(result) -> Dict[str, Any]:
     }
 
 
-def process_audit_result(
-    result, include_attributes: Optional[List[Union[str, AtlanField]]]
-) -> Dict[str, Any]:
+def process_audit_result(result) -> Dict[str, Any]:
     """
     Process a single audit result into a formatted dictionary.
 
     Args:
         result: Audit result object
-        include_attributes: List of attributes to include
 
     Returns:
         Formatted audit entry dictionary
     """
     try:
         # Extract basic audit information
         audit_entry = extract_basic_audit_info(result)
-
-        # Add requested attributes if specified
-        if include_attributes:
-            requested_attrs = result.detail.attributes.dict(
-                exclude_unset=True, by_alias=True
-            )
-            audit_entry.update(requested_attrs)
+        updates = result.detail.dict(exclude_unset=True)
+        audit_entry.update(updates)
 
         return audit_entry
 
@@ -184,21 +151,3 @@ def create_sort_item(sort_order: str) -> SortItem:
         "created",
         order=SortOrder.DESCENDING if sort_order == "DESC" else SortOrder.ASCENDING,
     )
-
-
-def convert_attributes_to_camel_case(
-    include_attributes: List[Union[str, AtlanField]],
-) -> List[Union[str, AtlanField]]:
-    """
-    Convert string attributes from snake_case to camelCase.
-
-    Args:
-        include_attributes: List of attributes to convert
-
-    Returns:
-        List with string attributes converted to camelCase
-    """
-    return [
-        snake_to_camel(attr) if isinstance(attr, str) else attr
-        for attr in include_attributes
-    ]