create_collection

Create a new collection in Google Threat Intelligence to organize and analyze Indicators of Compromise (IOCs) like domains, files, IP addresses, or URLs for threat investigation.

Instructions

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.

Input Schema

TableJSON Schema

Name	Required	Description	Default
`name`	Yes
`description`	Yes
`iocs`	Yes
`private`	No
`api_key`	No

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

Google Threat Intelligence MCP Server