Kaspersky OpenTIP MCP Server

Instructions

Implementation Reference

• opentip-mcp/opentip.py:142-153 (handler)

  The handler function for the 'search_ip' MCP tool. Validates the input IP address using ip_pattern regex, constructs request parameters, and delegates the API call to the opentip_request helper using the Endpoints.search_ip path.
  async def search_ip(ip: str) -> dict[str, Any] | None: """Get threat intelligence data about an IP address Args: ip: IPv4 address that you want to investigate """ if not ip_pattern.match(ip): return {"result": "error", "error_message": "Invalid IP address format. Please provide a valid IPv4 address."} params = {"request": ip} return await opentip_request(Endpoints.search_ip, "get", params)
• opentip-mcp/opentip.py:134-141 (registration)

  The @mcp.tool decorator registers the search_ip function as an MCP tool, providing a description and ToolAnnotations for metadata like title and hints.
  @mcp.tool( description="Get threat intelligence data about an IP address", annotations=ToolAnnotations( title="Investigate an IP", readOnlyHint=True, openWorldHint=True, ), )
• opentip-mcp/opentip.py:31-31 (schema)

  Regex pattern defining the schema for valid IPv4 addresses, used for input validation in the search_ip handler.
  ip_pattern = re.compile(r'^(\d{1,3}\.){3}\d{1,3}$')
• opentip-mcp/opentip.py:44-51 (helper)

  StrEnum defining API endpoints, including Endpoints.search_ip = 'search/ip' used by the search_ip handler.
  class Endpoints(StrEnum): search_hash = "search/hash" search_ip = "search/ip" search_domain = "search/domain" search_url = "search/url" analyze_file = "scan/file" get_analysis_results = "getresult/file"
• opentip-mcp/opentip.py:53-92 (helper)

  Helper function that performs the actual HTTP request to the OpenTIP API, handles errors, and is called by the search_ip handler.
  async def opentip_request( endpoint: str, request_type: RequestType = "get", params: Optional[dict[str, Any]] = None, content: Optional[bytes] = None, headers: Optional[dict[str, str]] = None, ) -> dict[str, Any]: """Make a request to the OpenTIP API with proper error handling.""" headers = headers or {} headers = { "user-agent": "opentip-mcp-client", "x-api-key": OPENTIP_API_KEY, **headers } async with httpx.AsyncClient() as client: try: url = f"{OPENTIP_API_BASE}{endpoint}" if request_type == "get": response = await client.get( url, headers=headers, params=params, timeout=OPENTIP_API_TIMEOUT ) elif request_type == "post": response = await client.post( url, headers=headers, params=params, content=content, timeout=OPENTIP_API_TIMEOUT ) response.raise_for_status() return response.json() except httpx.HTTPStatusError as e: if e.response.status_code == 400: return {"result": "error", "error_message": "Invalid parameters. Please check your input and try again."} elif e.response.status_code == 401: return {"result": "error", "error_message": "Authentication failed. Please ensure that you have provided the correct credentials and try again."} elif e.response.status_code == 403: return {"result": "error", "error_message": "Quota or request limit exceeded. Check your quota and limits and try again."} else: return {"result": "error", "error_message": str(e)} except Exception as e: # noqa return {"result": "error", "error_message": str(e)}