feat: Add announcement support to update_assets tool

modelcontextprotocol/pyproject.toml

@@ -18,7 +18,7 @@ classifiers = [
 dependencies = [
     "fastmcp==2.13.0.2",
     "pyatlan>=6.0.1",
-    "uvicorn>=0.35.0"
+    "uvicorn>=0.38.0"
 ]
 
 [project.scripts]

modelcontextprotocol/server.py

@@ -2,6 +2,7 @@
 import json
 import os
 from typing import Any, Dict, List
+from pydantic import ValidationError
 from fastmcp import FastMCP
 from tools import (
     search_assets,
@@ -16,6 +17,7 @@
     CertificateStatus,
     UpdatableAsset,
     TermOperations,
+    Announcement,
 )
 from pyatlan.model.lineage import LineageDirection
 from utils.parameters import (
@@ -480,7 +482,7 @@ def update_assets_tool(
             Can be a single UpdatableAsset or a list of UpdatableAsset objects.
             For asset of type_name=AtlasGlossaryTerm or type_name=AtlasGlossaryCategory, each asset dictionary MUST include a "glossary_guid" key which is the GUID of the glossary that the term belongs to.
         attribute_name (str): Name of the attribute to update.
-            Supports "user_description", "certificate_status", "readme", and "term".
+            Supports "user_description", "certificate_status", "readme", "term", and "announcement".
         attribute_values (List[Union[str, Dict[str, Any]]]): List of values to set for the attribute.
             For certificateStatus, only "VERIFIED", "DRAFT", or "DEPRECATED" are allowed.
             For readme, the value must be a valid Markdown string.
@@ -602,6 +604,30 @@ def update_assets_tool(
                 "term_guids": ["term-guid-to-remove"]
             }]
         )
+
+        # Add warning announcement to an asset
+        update_assets_tool(
+            assets={
+                "guid": "abc-123",
+                "name": "sales_data",
+                "type_name": "Table",
+                "qualified_name": "default/snowflake/db/schema/sales_data"
+            },
+            attribute_name="announcement",
+            attribute_values=[{
+                "announcement_type": "WARNING",
+                "announcement_title": "Data Quality Issue",
+                "announcement_message": "Missing records for Q4 2024. ETL team investigating."
+            }]
+        )
+
+        # Remove announcement
+        update_assets_tool(
+            assets={"guid": "abc-123", ...},
+            attribute_name="announcement",
+            attribute_values=[None]  # or [{}]
+        )
+
     """
     try:
         # Parse JSON parameters
@@ -629,6 +655,22 @@ def update_assets_tool(
                 CertificateStatus(val) for val in parsed_attribute_values
             ]
 
+        elif attr_enum == UpdatableAttribute.ANNOUNCEMENT:
+            announcement_objects = []
+            for value in parsed_attribute_values:
+                if value and not isinstance(value, dict):
+                    return {
+                        "error": f"Announcement must be a dict, got {type(value).__name__}",
+                        "updated_count": 0,
+                    }
+                if value:
+                    if "announcement_type" in value:
+                        value["announcement_type"] = value["announcement_type"].upper()
+                    announcement_objects.append(Announcement(**value))
+                else:
+                    announcement_objects.append(None)
+            parsed_attribute_values = announcement_objects
+
         # Convert assets to UpdatableAsset objects
         if isinstance(parsed_assets, dict):
             updatable_assets = [UpdatableAsset(**parsed_assets)]
@@ -640,7 +682,7 @@ def update_assets_tool(
             attribute_name=attr_enum,
             attribute_values=parsed_attribute_values,
         )
-    except (json.JSONDecodeError, ValueError, TypeError) as e:
+    except (json.JSONDecodeError, ValueError, TypeError, ValidationError) as e:
         return {
             "error": f"Parameter parsing/conversion error: {str(e)}",
             "updated_count": 0,

modelcontextprotocol/tools/__init__.py

@@ -16,6 +16,8 @@
     Glossary,
     GlossaryCategory,
     GlossaryTerm,
+    Announcement,
+    AnnouncementType,
 )
 
 __all__ = [
@@ -34,4 +36,6 @@
     "Glossary",
     "GlossaryCategory",
     "GlossaryTerm",
+    "AnnouncementType",
+    "Announcement",
 ]

modelcontextprotocol/tools/assets.py

@@ -7,8 +7,10 @@
     CertificateStatus,
     TermOperation,
     TermOperations,
+    Announcement,
 )
 from pyatlan.model.assets import Readme, AtlasGlossaryTerm, AtlasGlossaryCategory
+from pyatlan.model.enums import AnnouncementType as AtlanAnnouncementType
 from pyatlan.model.fluent_search import CompoundQuery, FluentSearch
 
 # Initialize logging
@@ -28,11 +30,15 @@ def update_assets(
             Can be a single UpdatableAsset or a list of UpdatableAssets.
             For asset of type_name=AtlasGlossaryTerm or type_name=AtlasGlossaryCategory, each asset dictionary MUST include a "glossary_guid" key which is the GUID of the glossary that the term belongs to.
         attribute_name (UpdatableAttribute): Name of the attribute to update.
-            Supports userDescription, certificateStatus, readme, and term.
+            Supports userDescription, certificateStatus, readme, term, and announcement.
         attribute_values (List[Union[str, CertificateStatus, TermOperations]]): List of values to set for the attribute.
             For certificateStatus, only VERIFIED, DRAFT, or DEPRECATED are allowed.
             For readme, the value must be a valid Markdown string.
             For term, the value must be a TermOperations object with operation and term_guids.
+            For announcement, each value should be a dict with:
+                - announcement_type: "INFORMATION", "WARNING", or "ISSUE"
+                - announcement_title: Title of the announcement
+                - announcement_message: Message content
 
     Returns:
         Dict[str, Any]: Dictionary containing:
@@ -168,8 +174,45 @@ def update_assets(
                     error_msg = f"Error updating terms on asset {updatable_asset.qualified_name}: {str(e)}"
                     logger.error(error_msg)
                     result["errors"].append(error_msg)
+            elif attribute_name == UpdatableAttribute.ANNOUNCEMENT:
+                announcement_data = attribute_values[index]
+                if not announcement_data:
+                    try:
+                        asset.remove_announcement()
+                        assets.append(asset)
+                        result["updated_count"] += 1
+                        logger.info(
+                            f"Successfully removed announcement from asset: {updatable_asset.qualified_name}"
+                        )
+                    except Exception as e:
+                        error_msg = f"Error removing announcement from asset {updatable_asset.qualified_name}: {str(e)}"
+                        logger.error(error_msg)
+                        result["errors"].append(error_msg)
+                else:
+                    if not isinstance(announcement_data, Announcement):
+                        error_msg = f"Announcement value must be an Announcement object for asset {updatable_asset.qualified_name}"
+                        logger.error(error_msg)
+                        result["errors"].append(error_msg)
+                        continue
+
+                    try:
+                        asset.announcement_type = AtlanAnnouncementType[
+                            announcement_data.announcement_type.value
+                        ]
+                        asset.announcement_title = announcement_data.announcement_title
+                        asset.announcement_message = (
+                            announcement_data.announcement_message
+                        )
+                        assets.append(asset)
+                        result["updated_count"] += 1
+                        logger.info(
+                            f"Successfully updated announcement on asset: {updatable_asset.qualified_name}"
+                        )
+                    except (KeyError, AttributeError, TypeError, Exception) as e:
+                        error_msg = f"Error setting announcement on asset {updatable_asset.qualified_name}: {str(e)}"
+                        logger.error(error_msg)
+                        result["errors"].append(error_msg)
             else:
-                # Regular attribute update flow
                 setattr(asset, attribute_name.value, attribute_values[index])
                 assets.append(asset)
 

modelcontextprotocol/tools/models.py

@@ -19,6 +19,7 @@ class UpdatableAttribute(str, Enum):
     CERTIFICATE_STATUS = "certificate_status"
     README = "readme"
     TERM = "term"
+    ANNOUNCEMENT = "announcement"
 
 
 class TermOperation(str, Enum):
@@ -74,3 +75,19 @@ class GlossaryTerm(BaseModel):
     user_description: Optional[str] = None
     certificate_status: Optional[CertificateStatus] = None
     category_guids: Optional[List[str]] = None
+
+
+class AnnouncementType(str, Enum):
+    """Enum for announcement types."""
+
+    INFORMATION = "INFORMATION"
+    WARNING = "WARNING"
+    ISSUE = "ISSUE"
+
+
+class Announcement(BaseModel):
+    """Class representing an announcement."""
+
+    announcement_type: AnnouncementType
+    announcement_title: str
+    announcement_message: str