Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/collections.py:428-461 (handler)

  The main implementation of create_collection tool. It's decorated with @server.tool() for registration and creates a new collection in Google Threat Intelligence by posting to the /collections endpoint with the provided name, description, IOCs, and privacy setting.

  @server.tool()
  async def create_collection(
      name: str,
      description: str,
      iocs: typing.List[str],
      ctx: Context,
      private: bool = True,
      api_key: str = None,
  ) -> typing.Dict[str, typing.Any]:
    """Creates a new collection in Google Threat Intelligence.
        Ask for the collection's privacy (public or private) if the user doesn't specify.
  
    Args:
      name (required): The name of the collection.
      description (required): A description of the collection.
      iocs (required): Indicators of Compromise (IOCs) to include in the
        collection. The items in the list can be domains, files, ip_addresses, or urls.
        At least one IOC must be provided.
      private: Indicates whether the collection should be private.
    Returns:
      A dictionary representing the newly created collection.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      collection_data = {
          "data": {
              "attributes": {"name": name, "description": description, "private": private},
              "type": "collection",
              "raw_items": ", ".join(iocs),
          }
      }
      
      res = await client.post_async("/collections", json_data=collection_data)
      data = await res.json_async()
    return utils.sanitize_response(data["data"])

• gti_mcp/tools/collections.py:24-54 (schema)

  Schema-related constants defining collection types, relationships, and excluded attributes. These define the valid structure and relationships for collections in the system.

  COLLECTION_RELATIONSHIPS = [
      "associations",
      "attack_techniques",
      "domains",
      "files",
      "ip_addresses",
      "urls",
      "threat_actors",
      "malware_families",
      "software_toolkits",
      "campaigns",
      "vulnerabilities",
      "reports",
      "suspected_threat_actors",
      "hunting_rulesets",
  ]
  
  COLLECTION_KEY_RELATIONSHIPS = [
      "associations",
  ]
  COLLECTION_EXCLUDED_ATTRS = ",".join(["aggregations"])
  
  COLLECTION_TYPES = {
      "threat-actor",
      "malware-family",
      "campaign",
      "report",
      "software-toolkit",
      "vulnerability",
      "collection",
  }

• gti_mcp/server.py:67-74 (registration)

  The FastMCP server instance creation and tool loading. The server is created with dependencies, then all tools (including create_collection) are imported from gti_mcp.tools which triggers the @server.tool() decorators.

  server = FastMCP(
      "Google Threat Intelligence MCP server",
      dependencies=["vt-py"],
      stateless_http=stateless)
  
  # Load tools.
  from gti_mcp.tools import *

• gti_mcp/server.py:56-64 (helper)

  Context manager that provides a VirusTotal client instance. Used by create_collection to get a VT client for making API requests.

  @asynccontextmanager
  async def vt_client(ctx: Context, api_key: str = None) -> AsyncIterator[vt.Client]:
    """Provides a vt.Client instance for the current request."""
    client = vt_client_factory(ctx, api_key)
  
    try:
      yield client
    finally:
      await client.close_async()

• gti_mcp/utils.py:119-138 (helper)

  Utility function that recursively removes empty dictionaries and lists from API responses. Used by create_collection to sanitize the response before returning it.

  def sanitize_response(data: typing.Any) -> typing.Any:
    """Removes empty dictionaries and lists recursively from a response."""
    if isinstance(data, dict):
      sanitized_dict = {}
      for key, value in data.items():
        sanitized_value = sanitize_response(value)
        if sanitized_value is not None:
          sanitized_dict[key] = sanitized_value
      return sanitized_dict
    elif isinstance(data, list):
      sanitized_list = []
      for item in data:
        sanitized_item = sanitize_response(item)
        if sanitized_item is not None:
          sanitized_list.append(sanitized_item)
      return sanitized_list
    elif isinstance(data, str):
      return data if data else None
    else:
      return data