Threat.Zone MCP Server

Instructions

Implementation Reference

• src/threatzone_mcp/server.py:238-252 (handler)

  The main handler function for the 'scan_url' tool. Decorated with @app.tool, which registers it as an MCP tool in FastMCP. It sends a POST request to the ThreatZone API to scan the provided URL.
  @app.tool async def scan_url(url: str, is_public: bool = False) -> Dict[str, Any]: """ Analyze a URL for threats and malicious content. Args: url: The URL to analyze is_public: Whether the scan results should be public """ data = { "url": url, "isPublic": is_public } return await get_client().post("/public-api/scan/url", data=data)
• src/threatzone_mcp/server.py:238-238 (registration)

  The @app.tool decorator registers the scan_url function as an MCP tool.
  @app.tool
• src/threatzone_mcp/server.py:102-110 (helper)

  Helper function get_client() used by scan_url to obtain the API client instance for making requests.
  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-97 (helper)

  APIClient class providing the post method used by scan_url to interact with the ThreatZone API, including error handling.
  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)