Threat.Zone MCP Server

Instructions

Implementation Reference

• src/threatzone_mcp/server.py:254-341 (handler)

  Primary handler for the 'scan_file_sandbox' tool. Decorated with @app.tool for FastMCP registration. Implements file upload to ThreatZone sandbox API, constructs detailed analysis configuration using metafields, supports numerous sandbox parameters like environment, timeout, network settings, etc., and returns the API submission response.
  @app.tool async def scan_file_sandbox( file_path: str, is_public: bool = False, entrypoint: Optional[str] = None, password: Optional[str] = None, environment: str = "w10_x64", timeout: int = 180, work_path: str = "desktop", mouse_simulation: bool = True, https_inspection: bool = False, internet_connection: bool = False, raw_logs: bool = False, snapshot: bool = False, sleep_evasion: bool = False, smart_tracing: bool = False, dump_collector: bool = False, open_in_browser: bool = False, extension_check: bool = True, modules: Optional[List[str]] = None, auto_config: bool = False ) -> Dict[str, Any]: """ Submit a file for advanced sandbox analysis with detailed configuration. Args: file_path: Path to the file to analyze is_public: Whether the scan results should be public (default: False) entrypoint: File to execute within archive (if applicable) password: Password for archive files (if applicable) environment: Analysis environment - w7_x64, w10_x64, w11_x64, macos, android, linux (default: w10_x64) timeout: Analysis timeout in seconds - 60, 120, 180, 240, 300 (default: 180) work_path: Working directory - desktop, root, %AppData%, windows, temp (default: desktop) mouse_simulation: Enable mouse simulation (default: True) https_inspection: Enable HTTPS inspection (default: False) internet_connection: Enable internet connection (default: False) raw_logs: Include raw logs (default: False) snapshot: Take VM snapshots (default: False) sleep_evasion: Enable sleep evasion techniques (default: False) smart_tracing: Enable smart tracing (default: False) dump_collector: Enable dump collection (default: False) open_in_browser: Open files in browser (default: False) extension_check: Perform extension check (default: True) modules: Analysis modules to use, e.g., ["csi", "cdr"] (default: None) auto_config: Use automatic configuration (default: False) """ if not Path(file_path).exists(): raise ThreatZoneError(f"File not found: {file_path}") # Build the analyze configuration analyze_config = [ {"metafieldId": "environment", "value": environment}, {"metafieldId": "private", "value": not is_public}, {"metafieldId": "timeout", "value": timeout}, {"metafieldId": "work_path", "value": work_path}, {"metafieldId": "mouse_simulation", "value": mouse_simulation}, {"metafieldId": "https_inspection", "value": https_inspection}, {"metafieldId": "internet_connection", "value": internet_connection}, {"metafieldId": "raw_logs", "value": raw_logs}, {"metafieldId": "snapshot", "value": snapshot}, {"metafieldId": "sleep_evasion", "value": sleep_evasion}, {"metafieldId": "smart_tracing", "value": smart_tracing}, {"metafieldId": "dump_collector", "value": dump_collector}, {"metafieldId": "open_in_browser", "value": open_in_browser} ] # Prepare form data data = { "analyzeConfig": json.dumps(analyze_config), "extensionCheck": str(extension_check).lower() } if entrypoint: data["entrypoint"] = entrypoint if password: data["password"] = password if modules: data["modules"] = ",".join(modules) # Build URL with auto parameter url = f"/public-api/scan/sandbox?auto={str(auto_config).lower()}" files = {"file": open(file_path, "rb")} try: return await get_client().post(url, data=data, files=files) finally: files["file"].close()
• src/threatzone_mcp/server.py:343-369 (helper)

  Helper wrapper tool 'scan_file_sandbox_simple' that provides a simplified interface by calling the main scan_file_sandbox with auto_config=True and defaults.
  @app.tool async def scan_file_sandbox_simple( file_path: str, is_public: bool = False, entrypoint: Optional[str] = None, password: Optional[str] = None ) -> Dict[str, Any]: """ Submit a file for simple sandbox analysis using default settings. This is a simplified version of scan_file_sandbox with default configurations. Use scan_file_sandbox for advanced configuration options. Args: file_path: Path to the file to analyze is_public: Whether the scan results should be public (default: False) entrypoint: File to execute within archive (if applicable) password: Password for archive files (if applicable) """ return await scan_file_sandbox( file_path=file_path, is_public=is_public, entrypoint=entrypoint, password=password, auto_config=True # Use automatic configuration for simplicity )
• src/threatzone_mcp/server.py:102-110 (helper)

  Utility function to lazily initialize the APIClient used by scan_file_sandbox for making HTTP requests to ThreatZone API.
  def get_client(): """Get or create the API client.""" global client if client is None: if not API_KEY: raise ThreatZoneError("THREATZONE_API_KEY environment variable is required") client = APIClient(API_KEY) return client
• src/threatzone_mcp/server.py:29-98 (helper)

  APIClient class providing the post method with file upload support, crucial for the sandbox scan implementation.
  class APIClient: """HTTP client for Threat.Zone API.""" def __init__(self, api_key: str, base_url: str = API_BASE_URL): self.api_key = api_key self.base_url = base_url self.headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } async def get(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """Make GET request to API.""" async with httpx.AsyncClient() as client: response = await client.get( f"{self.base_url}{endpoint}", headers=self.headers, params=params ) await self._handle_response(response) return response.json() async def post(self, endpoint: str, data: Optional[Dict[str, Any]] = None, files: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """Make POST request to API.""" async with httpx.AsyncClient() as client: headers = {"Authorization": f"Bearer {self.api_key}"} if files: # For file uploads, don't set Content-Type response = await client.post( f"{self.base_url}{endpoint}", headers=headers, data=data, files=files ) else: response = await client.post( f"{self.base_url}{endpoint}", headers=self.headers, json=data ) await self._handle_response(response) return response.json() async def download(self, endpoint: str) -> bytes: """Download file from API.""" async with httpx.AsyncClient() as client: response = await client.get( f"{self.base_url}{endpoint}", headers=self.headers ) await self._handle_response(response) return response.content async def _handle_response(self, response: httpx.Response) -> None: """Handle API response errors.""" if response.status_code == 401: raise ThreatZoneError("Authentication failed. Check your API key.") elif response.status_code == 404: raise ThreatZoneError("Resource not found.") elif response.status_code == 422: raise ThreatZoneError("Invalid request parameters.") elif response.status_code >= 400: try: error_data = response.json() error_msg = error_data.get("message", f"API error: {response.status_code}") except: error_msg = f"API error: {response.status_code}" raise ThreatZoneError(error_msg)