Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/intelligence.py:27-69 (handler)

  The main handler function for search_iocs tool. Searches Indicators of Compromise (IOC) in Google Threat Intelligence platform. Uses query, limit, order_by, and api_key parameters. Calls vt_client and uses helper utilities to fetch and sanitize results.

  @server.tool()
  async def search_iocs(query: str, ctx: Context, limit: int = 10, order_by: str = "last_submission_date-", api_key: str = None) -> typing.List[typing.Dict[str, typing.Any]]:
    """Search Indicators of Compromise (IOC) in the Google Threat Intelligence platform.
  
    You can search by for different IOC types using the `entity` modifier. Below, the different IOC types and the supported orders:
  
      | Entity type   | Supported orders | Default order |
      | ------------- | ---------------- | ------------- |
      | file          | first_submission_date, last_submission_date, positives, times_submitted, size	    | last_submission_date- |
      | url           | first_submission_date, last_submission_date, positives, times_submitted, status   | last_submission_date- |
      | domain        | creation_date, last_modification_date, last_update_date, positives                | last_modification_date- |
      | ip            | ip, last_modification_date, positives                                             | last_modification_date- |
  
    Note: The `entity` modifier can only be used ONCE per query.
  
    You can find all available modifers at:
      - Files: https://gtidocs.virustotal.com/docs/file-search-modifiers
      - URLs: https://gtidocs.virustotal.com/docs/url-search-modifiers
      - Domains: https://gtidocs.virustotal.com/docs/domain-search-modifiers
      - IP Addresses: https://gtidocs.virustotal.com/docs/ip-address-search-modifiers
  
    With integer modifers, use the `-` and `+` characters to indicate:
      - Greater than: `p:60+`
      - Less than: `p:60-`
      - Equal to: `p:60`
  
    Args
      query (required): Search query to find IOCs.
      limit: Limit the number of IoCs to retrieve. 10 by default.
      order_by: Order the results. "last_submission_date-" by default.
  
    Returns:
      List of Indicators of Compromise (IoCs).
    """
    async with vt_client(ctx, api_key=api_key) as client:
      res = await utils.consume_vt_iterator(
          client,
          "/intelligence/search",
          params={
              "query": query,
              "order": order_by},
          limit=limit)
    return utils.sanitize_response([o.to_dict() for o in res])

• gti_mcp/tools/intelligence.py:27-27 (registration)

  Registration of search_iocs tool via @server.tool() decorator from FastMCP framework. The decorator automatically registers the function as an available MCP tool.

  @server.tool()

• gti_mcp/utils.py:20-26 (helper)

  Helper function consume_vt_iterator used by search_iocs to iterate through VirusTotal API results and return them as a list of objects.

  async def consume_vt_iterator(
      vt_client: vt.Client, endpoint: str, params: dict | None = None, limit: int = 10):
    """Consumes a vt.Iterator iterator and return the list of objects."""
    res = []
    async for obj in vt_client.iterator(endpoint, params=params, limit=limit):
      res.append(obj)
    return res

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

  Helper function sanitize_response used by search_iocs to clean up the API response by recursively removing empty dictionaries and lists.

  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