Google Threat Intelligence MCP Server

Instructions

Implementation Reference

• gti_mcp/tools/files.py:251-269 (handler)

  The main handler function for the analyse_file tool. It's decorated with @server.tool() which registers it as an MCP tool. The function uploads a file to VirusTotal for analysis, waits for completion, and returns the analysis report.

  @server.tool()
  async def analyse_file(file_path: str, ctx: Context, api_key: str = None):
    """Upload and analyse the file in VirusTotal.
  
    The file will be uploaded to VirusTotal and shared with the community.
  
    Args:
      file_path (required): Path to the file for analysis. Use absolute path.
    Returns:
      The analysis report.
    """
    async with vt_client(ctx, api_key=api_key) as client:
      with open(file_path, "rb") as f:    
        analysis = await client.scan_file_async(file=f)
        logging.info(f"File {file_path} uploaded.")
  
      res = await client.wait_for_analysis_completion(analysis)
      logging.info(f"Analysis has completed with ID %s", res.id)
      return utils.sanitize_response(res.to_dict())

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

  The vt_client async context manager that provides a VirusTotal client instance. It handles API key resolution and ensures proper cleanup of the client connection.

  @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)

  The sanitize_response utility function used by analyse_file to remove empty dictionaries and lists recursively from the API 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

• gti_mcp/server.py:40-51 (helper)

  The vt_client_factory function that creates VirusTotal client instances. It prioritizes the api_key argument over the VT_APIKEY environment variable.

  def _vt_client_factory(ctx: Context, api_key: str = None) -> vt.Client:
      # Prioritize the passed argument
      if not api_key:
          api_key = os.getenv("VT_APIKEY")
      
      # Try to get from context if not in env (placeholder for future ctx inspection)
      # if not api_key and ctx and hasattr(ctx, 'init_options'):
      #     api_key = ctx.init_options.get('vtApiKey')
      
      if not api_key:
          raise ValueError("VT API Key is required. Please provide it as an argument 'api_key' or set VT_APIKEY environment variable.")
      return vt.Client(api_key)

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

  The FastMCP server instance creation. Tools are registered via the @server.tool() decorator, which is applied to analyse_file.

  server = FastMCP(
      "Google Threat Intelligence MCP server",
      dependencies=["vt-py"],
      stateless_http=stateless)