feat(servers): add FastMCP tool annotations across SecOps, SOAR, GTI, and SCC (fixes #320)

server/gti/README.md

@@ -6,6 +6,13 @@ Threat Intelligence suite.
 
 ## Features
 
+### Tool Annotations & Safety Hints
+
+All tools exposed by the Google Threat Intelligence MCP server declare Model Context Protocol `ToolAnnotations` metadata to guide autonomous clients (such as Gemini CLI, Google ADK, Claude, and Cursor) regarding execution safety:
+
+- **Read-Only Tools (`readOnlyHint: true`)**: Threat reporting, entity lookups, IOC searches, domain and IP intelligence, threat profiles, and ruleset queries (32 tools total).
+- **Additive / Mutating Tools (`readOnlyHint: false, destructiveHint: false`)**: Operations that create or update threat collections or request file analysis (`create_collection`, `update_collection_attributes`, `update_iocs_in_collection`, `analyse_file`).
+
 ### Collections (Threats)
 
 - **`get_collection_report(id)`**: Retrieves a specific collection report by its ID (e.g., `report--<hash>`, `threat-actor--<hash>`).

server/gti/gti_mcp/tools/collections.py

@@ -54,7 +54,7 @@
 }
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_collection_report(id: str, ctx: Context) -> typing.Dict[str, typing.Any]:
   """At Google Threat Intelligence, threats are modeled as "collections". This tool retrieves them from the platform.
 
@@ -83,7 +83,7 @@ async def get_collection_report(id: str, ctx: Context) -> typing.Dict[str, typin
   return res
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_entities_related_to_a_collection(
     id: str, relationship_name: str, ctx: Context, limit: int = 10, descriptors_only: bool = True
 ) -> typing.List[typing.Dict[str, typing.Any]]:
@@ -173,7 +173,7 @@ async def _search_threats_by_collection_type(
   return utils.sanitize_response([o.to_dict() for o in res])
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def search_threats(
     ctx: Context,
     query: str,
@@ -236,7 +236,7 @@ async def search_threats(
   return utils.sanitize_response([o.to_dict() for o in res])
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def search_campaigns(
     query: str, ctx: Context, limit: int = 10, order_by: str = "relevance-"
 ) -> typing.List[typing.Dict[str, typing.Any]]:
@@ -259,7 +259,7 @@ async def search_campaigns(
   return res
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def search_threat_actors(
     query: str, ctx: Context, limit: int = 10, order_by: str = "relevance-"
 ) -> typing.List[typing.Dict[str, typing.Any]]:
@@ -282,7 +282,7 @@ async def search_threat_actors(
   return res
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def search_malware_families(
     query: str, ctx: Context, limit: int = 10, order_by: str = "relevance-"
 ) -> typing.List[typing.Dict[str, typing.Any]]:
@@ -305,7 +305,7 @@ async def search_malware_families(
   return res
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def search_software_toolkits(
     query: str, ctx: Context, limit: int = 10, order_by: str = "relevance-"
 ) -> typing.List[typing.Dict[str, typing.Any]]:
@@ -328,7 +328,7 @@ async def search_software_toolkits(
   return res
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def search_threat_reports(
     query: str, ctx: Context, limit: int = 10, order_by: str = "relevance-"
 ) -> typing.List[typing.Dict[str, typing.Any]]:
@@ -353,7 +353,7 @@ async def search_threat_reports(
   return res
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def search_vulnerabilities(
     query: str, ctx: Context, limit: int = 10, order_by: str = "relevance-"
 ) -> typing.List[typing.Dict[str, typing.Any]]:
@@ -376,7 +376,7 @@ async def search_vulnerabilities(
   return res
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_collection_timeline_events(id: str, ctx: Context):
   """Retrieves timeline events from the given collection, when available.
 
@@ -401,7 +401,7 @@ async def get_collection_timeline_events(id: str, ctx: Context):
   return utils.sanitize_response(data.get("data", []))
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_collection_mitre_tree(id: str, ctx: Context) -> typing.Dict:
   """Retrieves the Mitre tactics and techniques associated with a threat.
 
@@ -420,7 +420,12 @@ async def get_collection_mitre_tree(id: str, ctx: Context) -> typing.Dict:
   return utils.sanitize_response(data.get("data", {}))
 
 
-@server.tool()
+@server.tool(
+    annotations={
+        "readOnlyHint": False,
+        "destructiveHint": False,
+    }
+)
 async def create_collection(
     name: str,
     description: str,
@@ -455,7 +460,12 @@ async def create_collection(
   return utils.sanitize_response(data["data"])
 
 
-@server.tool()
+@server.tool(
+    annotations={
+        "readOnlyHint": False,
+        "destructiveHint": False,
+    }
+)
 async def update_collection_attributes(
     id: str,
     ctx: Context,
@@ -483,7 +493,12 @@ async def update_collection_attributes(
   return utils.sanitize_response(data["data"])
 
 
-@server.tool()
+@server.tool(
+    annotations={
+        "readOnlyHint": False,
+        "destructiveHint": False,
+    }
+)
 async def update_iocs_in_collection(
     id: str,
     ctx: Context,
@@ -535,7 +550,7 @@ async def update_iocs_in_collection(
     return 'Sucesssfully updated collection' if status == 200 else 'Error updating collection'
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_collection_feature_matches(
     collection_id: str,
     feature_type: str,
@@ -636,7 +651,7 @@ async def get_collection_feature_matches(
     return utils.sanitize_response(data.get("data", []))
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_collections_commonalities(collection_id: str, ctx: Context) -> str:
   """Retrieve the common characteristics or features (attributes / relationships) of the indicators of compromise (IoC) within a collection, identified by its ID.
   Args:
@@ -706,7 +721,7 @@ async def _get_sigma_rule_details(ctx: Context, rule: dict, rule_type: str) -> t
     logging.exception("Error fetching Sigma ruleset %s: %s", ruleset_id, e)
     return {"error": f"Error fetching Sigma ruleset {ruleset_id}: {e}"}
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_collection_rules(collection_id: str, ctx: Context, top_n: int = 4, rule_types: typing.List[str] = None) -> typing.Union[typing.List[typing.Dict[str, typing.Any]], typing.Dict[str, str]]:
   """Retrieve top N community rules and all curated hunting rules for a specific collection.
 

server/gti/gti_mcp/tools/files.py

@@ -85,7 +85,7 @@
 ]
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_file_report(hash: str, ctx: Context) -> typing.Dict[str, typing.Any]:
   """Get a comprehensive file analysis report using its hash (MD5/SHA-1/SHA-256).
 
@@ -107,7 +107,7 @@ async def get_file_report(hash: str, ctx: Context) -> typing.Dict[str, typing.An
   return utils.sanitize_response(res)
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_entities_related_to_a_file(
     hash: str, relationship_name: str, descriptors_only: bool, ctx: Context, limit: int = 10, 
 ) -> list[dict[str, typing.Any]]:
@@ -190,7 +190,7 @@ async def get_entities_related_to_a_file(
     return utils.sanitize_response(res.get(relationship_name, []))
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_file_behavior_report(
     file_behaviour_id: str, ctx: Context
 ) -> typing.Dict[str, typing.Any]:
@@ -225,7 +225,7 @@ async def get_file_behavior_report(
   return utils.sanitize_response(res)
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_file_behavior_summary(hash: str, ctx: Context) -> typing.Dict[str, typing.Any]:
   """Retrieve a summary of all the file behavior reports from all the sandboxes.
 
@@ -248,7 +248,12 @@ async def get_file_behavior_summary(hash: str, ctx: Context) -> typing.Dict[str,
   return utils.sanitize_response(res["data"])
 
 
-@server.tool()
+@server.tool(
+    annotations={
+        "readOnlyHint": False,
+        "destructiveHint": False,
+    }
+)
 async def analyse_file(file_path: str, ctx: Context):
   """Upload and analyse the file in VirusTotal.
 
@@ -268,7 +273,7 @@ async def analyse_file(file_path: str, ctx: Context):
     logging.info(f"Analysis has completed with ID %s", res.id)
     return utils.sanitize_response(res.to_dict())
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def search_digital_threat_monitoring(
     query: str,
     ctx: Context,

server/gti/gti_mcp/tools/intelligence.py

@@ -24,7 +24,7 @@
 ]
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def search_iocs(query: str, ctx: Context, limit: int = 10, order_by: str = "last_submission_date-") -> typing.List[typing.Dict[str, typing.Any]]:
   """Search Indicators of Compromise (IOC) in the Google Threat Intelligence platform.
 
@@ -69,7 +69,7 @@ async def search_iocs(query: str, ctx: Context, limit: int = 10, order_by: str =
   return utils.sanitize_response([o.to_dict() for o in res])
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_hunting_ruleset(ruleset_id: str, ctx: Context) -> typing.Dict[str, typing.Any]:
   """Get a Hunting Ruleset object from Google Threat Intelligence.
 
@@ -103,7 +103,7 @@ async def get_hunting_ruleset(ruleset_id: str, ctx: Context) -> typing.Dict[str,
   return utils.sanitize_response(res)
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_entities_related_to_a_hunting_ruleset(
     ruleset_id: str, relationship_name: str, ctx: Context, limit: int = 10
 ) -> list[dict[str, typing.Any]]:

server/gti/gti_mcp/tools/netloc.py

@@ -88,7 +88,7 @@
 ]
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_domain_report(domain: str, ctx: Context) -> typing.Dict[str, typing.Any]:
   """Get a comprehensive domain analysis report from Google Threat Intelligence.
 
@@ -108,7 +108,7 @@ async def get_domain_report(domain: str, ctx: Context) -> typing.Dict[str, typin
   return utils.sanitize_response(res)
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_entities_related_to_a_domain(
     domain: str, relationship_name: str, descriptors_only: bool, ctx: Context, limit: int = 10
 ) -> list[dict[str, typing.Any]]:
@@ -174,7 +174,7 @@ async def get_entities_related_to_a_domain(
   return utils.sanitize_response(res.get(relationship_name, []))
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_ip_address_report(ip_address: str, ctx: Context) -> typing.Dict[str, typing.Any]:
   """Get a comprehensive IP Address analysis report from Google Threat Intelligence.
 
@@ -193,7 +193,7 @@ async def get_ip_address_report(ip_address: str, ctx: Context) -> typing.Dict[st
   return utils.sanitize_response(res)
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_entities_related_to_an_ip_address(
     ip_address: str, relationship_name: str, descriptors_only: bool, ctx: Context, limit: int = 10
 ) -> list[dict[str, typing.Any]]:

server/gti/gti_mcp/tools/threat_profiles.py

@@ -19,7 +19,7 @@
 from ..server import server, vt_client
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def list_threat_profiles(
     ctx: Context, limit: int = 10
 ) -> typing.List[typing.Dict[str, typing.Any]]:
@@ -51,7 +51,7 @@ async def list_threat_profiles(
   return utils.sanitize_response([o.to_dict() for o in res])
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_threat_profile(
     profile_id: str, ctx: Context
 ) -> typing.Dict[str, typing.Any]:
@@ -99,7 +99,7 @@ async def get_threat_profile(
   return utils.sanitize_response(res)
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_threat_profile_recommendations(
     profile_id: str, ctx: Context, limit: int = 10
 ) -> typing.List[typing.Dict[str, typing.Any]]:
@@ -144,7 +144,7 @@ async def get_threat_profile_recommendations(
   return utils.sanitize_response(res.get('recommendations', []))
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_threat_profile_associations_timeline(
     profile_id: str, ctx: Context, limit: int = 10
 ) -> typing.List[typing.Dict[str, typing.Any]]:

server/gti/gti_mcp/tools/urls.py

@@ -69,7 +69,7 @@ def url_to_base64(url: str) -> str:
   return b.decode('utf-8').rstrip("=")
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_url_report(url: str, ctx: Context) -> typing.Dict[str, typing.Any]:
   """Get a comprehensive URL analysis report from Google Threat Intelligence.
 
@@ -90,7 +90,7 @@ async def get_url_report(url: str, ctx: Context) -> typing.Dict[str, typing.Any]
   return utils.sanitize_response(res)
 
 
-@server.tool()
+@server.tool(annotations={"readOnlyHint": True})
 async def get_entities_related_to_an_url(
     url: str, relationship_name: str, descriptors_only: bool, ctx: Context, limit: int = 10
 ) -> list[dict[str, typing.Any]]: