Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/collections.py:409-425 (handler)

  The complete implementation of get_collection_mitre_tree tool. This function retrieves MITRE tactics and techniques associated with a collection from the Google Threat Intelligence API. It includes the @server.tool() decorator for registration and handles API errors gracefully by returning error dictionaries.

  @server.tool()
  async def get_collection_mitre_tree(id: str, ctx: Context, api_key: str = None) -> typing.Dict:
    """Retrieves the Mitre tactics and techniques associated with a threat.
  
    Args:
      id (required): Collection identifiers.
    Return:
      A dictionary including the tactics and techniques associated to the given threat.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      resp = await client.get_async(f"/collections/{id}/mitre_tree")
      if resp.status != 200:
          error_json = await resp.json_async()
          error_info = error_json.get("error", {})
          return {"error": f"API Error: {error_info.get('message', 'Unknown error')}"}
      data = await resp.json_async()
    return utils.sanitize_response(data.get("data", {}))

• gti_mcp/tools/collections.py:410-425 (schema)

  Function signature with type hints defining input/output schema: takes an id (str), ctx (Context), and optional api_key (str), returns typing.Dict. This serves as the type definition for the tool's interface.

  async def get_collection_mitre_tree(id: str, ctx: Context, api_key: str = None) -> typing.Dict:
    """Retrieves the Mitre tactics and techniques associated with a threat.
  
    Args:
      id (required): Collection identifiers.
    Return:
      A dictionary including the tactics and techniques associated to the given threat.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      resp = await client.get_async(f"/collections/{id}/mitre_tree")
      if resp.status != 200:
          error_json = await resp.json_async()
          error_info = error_json.get("error", {})
          return {"error": f"API Error: {error_info.get('message', 'Unknown error')}"}
      data = await resp.json_async()
    return utils.sanitize_response(data.get("data", {}))

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

  The sanitize_response helper function used by get_collection_mitre_tree to clean API responses by removing empty dictionaries and lists recursively.

  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

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

  FastMCP server instance creation and tool loading. The server object is used by @server.tool() decorators to register tools, and line 73 imports all tools from gti_mcp.tools module.

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