feat: add additional scan details on push (#137)

Author: knielsen404

src/mcp_scan/MCPScanner.py

@@ -2,6 +2,7 @@
 import logging
 import os
 import re
+import time
 import traceback
 from collections import defaultdict
 from collections.abc import Callable
@@ -78,6 +79,7 @@ def __init__(
         additional_headers: dict | None = None,
         control_servers: list | None = None,
         skip_ssl_verify: bool = False,
+        scan_context: dict | None = None,
         **kwargs: Any,
     ):
         logger.info("Initializing MCPScanner")
@@ -97,6 +99,7 @@ def __init__(
         self.control_servers = control_servers
         self.verbose = verbose
         self.skip_ssl_verify = skip_ssl_verify
+        self.scan_context = scan_context if scan_context is not None else {}
         logger.debug(
             "MCPScanner initialized with timeout: %d, checks_per_server: %d", server_timeout, checks_per_server
         )
@@ -285,6 +288,7 @@ async def check_path(self, path_result: ScanPathResult) -> ScanPathResult:
 
     async def scan(self) -> list[ScanPathResult]:
         logger.info("Starting scan of %d paths", len(self.paths))
+        scan_start_time = time.perf_counter()
         if self.context_manager is not None:
             self.context_manager.disable()
 
@@ -309,6 +313,8 @@ async def scan(self) -> list[ScanPathResult]:
             verbose=self.verbose,
             skip_ssl_verify=self.skip_ssl_verify,
         )
+        self.scan_context["scan_time_milliseconds"] = (time.perf_counter() - scan_start_time) * 1000
+
         logger.debug("Result verified: %s", result_verified)
         logger.debug("Saving storage file")
         self.storage_file.save()

src/mcp_scan/cli.py

@@ -19,7 +19,7 @@
 from mcp_scan.MCPScanner import MCPScanner
 from mcp_scan.printer import print_scan_result
 from mcp_scan.Storage import Storage
-from mcp_scan.upload import upload
+from mcp_scan.upload import get_hostname, upload
 from mcp_scan.utils import parse_headers
 from mcp_scan.version import version_info
 from mcp_scan.well_known_clients import WELL_KNOWN_MCP_PATHS, client_shorthands_to_paths
@@ -712,6 +712,7 @@ def server(on_exit=None):
         parser.print_help()
         sys.exit(1)
 
+
 async def evo(args):
     """
     Pushes the scan results to the Evo API.
@@ -721,13 +722,15 @@ async def evo(args):
     3. Revokes the client_id
     """
 
-    rich.print(f"Go to https://app.snyk.io and select the tenant on the left nav bar. Copy the Tenant ID from the URL and paste it here: ")
+    rich.print(
+        "Go to https://app.snyk.io and select the tenant on the left nav bar. Copy the Tenant ID from the URL and paste it here: "
+    )
     tenant_id = input().strip()
-    rich.print(f"Paste the Authorization token from https://app.snyk.io/account (API Token -> KEY -> click to show): ")
+    rich.print("Paste the Authorization token from https://app.snyk.io/account (API Token -> KEY -> click to show): ")
     token = input().strip()
 
     push_key_url = f"https://api.snyk.io/hidden/tenants/{tenant_id}/mcp-scan/push-key?version=2025-08-28"
-    push_scan_url = f"https://api.snyk.io/hidden/mcp-scan/push?version=2025-08-28"
+    push_scan_url = "https://api.snyk.io/hidden/mcp-scan/push?version=2025-08-28"
 
     # create a client_id (shared secret)
     client_id = None
@@ -743,18 +746,18 @@ async def evo(args):
                 if not client_id:
                     rich.print(f"[bold red]Unexpected response[/bold red]: {data}")
                     return
-                rich.print(f"Client ID created")
+                rich.print("Client ID created")
     except Exception as e:
         rich.print(f"[bold red]Error calling Snyk API[/bold red]: {e}")
         return
 
     # Update the default scan args
-    args.control_servers=[
+    args.control_servers = [
         {
             "url": push_scan_url,
-            "identifier": None,
+            "identifier": get_hostname() or None,
             "opt_out": False,
-            "headers": [f"x-client-id:{client_id}"]
+            "headers": [f"x-client-id:{client_id}"],
         }
     ]
     await run_scan_inspect(mode="scan", args=args)
@@ -777,7 +780,12 @@ async def evo(args):
 
 
 async def run_scan_inspect(mode="scan", args=None):
-    async with MCPScanner(additional_headers=parse_headers(args.verification_H), **vars(args)) as scanner:
+    # Initialize scan_context dict that can be populated during scanning
+    scan_context = {"cli_version": version_info}
+
+    async with MCPScanner(
+        additional_headers=parse_headers(args.verification_H), scan_context=scan_context, **vars(args)
+    ) as scanner:
         if mode == "scan":
             result = await scanner.scan()
         elif mode == "inspect":
@@ -796,6 +804,7 @@ async def run_scan_inspect(mode="scan", args=None):
                 verbose=getattr(args, "verbose", False),
                 additional_headers=parse_headers(server_config["headers"]),
                 skip_ssl_verify=getattr(args, "skip_ssl_verify", False),
+                scan_context=scan_context,
             )
     return result
 

src/mcp_scan/models.py

@@ -375,3 +375,4 @@ class AnalysisServerResponse(BaseModel):
 class ScanPathResultsCreate(BaseModel):
     scan_path_results: list[ScanPathResult]
     scan_user_info: ScanUserInfo
+    scan_metadata: dict[str, Any] | None = None

src/mcp_scan/upload.py

@@ -62,6 +62,7 @@ async def upload(
     additional_headers: dict | None = None,
     max_retries: int = 3,
     skip_ssl_verify: bool = False,
+    scan_context: dict | None = None,
 ) -> None:
     """
     Upload the scan results to the control server with retry logic.
@@ -75,6 +76,7 @@ async def upload(
         additional_headers: Additional HTTP headers to send
         max_retries: Maximum number of retry attempts (default: 3)
         skip_ssl_verify: Whether to disable SSL certificate verification (default: False)
+        scan_context: Optional dict containing scan context metadata to include in upload
     """
     if not results:
         logger.info("No scan results to upload")
@@ -94,7 +96,11 @@ async def upload(
         result.client = get_client_from_path(result.path) or result.client or result.path
         results_with_servers.append(result)
 
-    payload = ScanPathResultsCreate(scan_path_results=results_with_servers, scan_user_info=user_info)
+    payload = ScanPathResultsCreate(
+        scan_path_results=results_with_servers,
+        scan_user_info=user_info,
+        scan_metadata=scan_context if scan_context else None,
+    )
 
     last_exception = None
     trace_configs = setup_aiohttp_debug_logging(verbose=verbose)