Kaspersky OpenTIP MCP Server

Instructions

Implementation Reference

• opentip-mcp/opentip.py:94-113 (handler)

  The main handler function for the 'search_hash' MCP tool. It validates the input hash using a regex pattern, prepares parameters, and calls the opentip_request helper to query the OpenTIP API. The @mcp.tool decorator registers it as an MCP tool with description and annotations.

  @mcp.tool(
      description="Get threat intelligence information about a file by hash (md5, sha1, sha256)",
      annotations=ToolAnnotations(
          title="Investigate a file by hash",
          readOnlyHint=True,
          openWorldHint=True,
      ),
  )
  async def search_hash(file_hash: str) -> dict[str, Any] | None:
      """Get threat intelligence information about a file by hash (md5, sha1, sha256)
  
      Args:
          file_hash: hash that you want to investigate
      """
  
      if not hash_pattern.match(file_hash):
          return {"result": "error", "error_message": "Invalid hash format. Please provide a valid md5, sha1, or sha256 hash."}
  
      params = {"request": file_hash}
      return await opentip_request(Endpoints.search_hash, "get", params)

• opentip-mcp/opentip.py:53-92 (helper)

  Shared helper function that performs HTTP requests to the OpenTIP API, handles authentication, errors, and returns JSON responses or error dicts. Called by search_hash with endpoint Endpoints.search_hash.

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

• opentip-mcp/opentip.py:44-51 (helper)

  StrEnum defining the API endpoint paths, including 'search_hash' used by the tool 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:30-30 (helper)

  Regex pattern used by search_hash for validating MD5 (32 hex), SHA1 (40 hex), SHA256 (64 hex) hashes.

  hash_pattern = re.compile(r'^(0x)?(?:[a-fA-F0-9]{32}|[a-fA-F0-9]{40}|[a-fA-F0-9]{64})$')