renamed to mcp-obsidian

src/mcp_obsidian/obsidian.py

@@ -0,0 +1,134 @@
+import requests
+import urllib.parse
+from typing import Any
+
+class Obsidian():
+    def __init__(
+            self, 
+            api_key: str,
+            protocol: str = 'https',
+            host: str = "127.0.0.1",
+            port: int = 27124,
+            verify_ssl: bool = False,
+        ):
+        self.api_key = api_key
+        self.protocol = protocol
+        self.host = host
+        self.port = port
+        self.verify_ssl = verify_ssl
+        self.timeout = (3, 6)
+
+    def get_base_url(self) -> str:
+        return f'{self.protocol}://{self.host}:{self.port}'
+    
+    def _get_headers(self) -> dict:
+        headers = {
+            'Authorization': f'Bearer {self.api_key}'
+        }
+        return headers
+
+    def _safe_call(self, f) -> Any:
+        try:
+            return f()
+        except requests.HTTPError as e:
+            error_data = e.response.json() if e.response.content else {}
+            code = error_data.get('errorCode', -1) 
+            message = error_data.get('message', '<unknown>')
+            raise Exception(f"Error {code}: {message}")
+        except requests.exceptions.RequestException as e:
+            raise Exception(f"Request failed: {str(e)}")
+
+    def list_files_in_vault(self) -> Any:
+        url = f"{self.get_base_url()}/vault/"
+        
+        def call_fn():
+            response = requests.get(url, headers=self._get_headers(), verify=self.verify_ssl, timeout=self.timeout)
+            response.raise_for_status()
+            
+            return response.json()['files']
+
+        return self._safe_call(call_fn)
+
+        
+    def list_files_in_dir(self, dirpath: str) -> Any:
+        url = f"{self.get_base_url()}/vault/{dirpath}/"
+        
+        def call_fn():
+            response = requests.get(url, headers=self._get_headers(), verify=self.verify_ssl, timeout=self.timeout)
+            response.raise_for_status()
+            
+            return response.json()['files']
+
+        return self._safe_call(call_fn)
+
+    def get_file_contents(self, filepath: str) -> Any:
+        url = f"{self.get_base_url()}/vault/{filepath}"
+    
+        def call_fn():
+            response = requests.get(url, headers=self._get_headers(), verify=self.verify_ssl, timeout=self.timeout)
+            response.raise_for_status()
+            
+            return response.text
+
+        return self._safe_call(call_fn)
+    
+    def search(self, query: str, context_length: int = 100) -> Any:
+        url = f"{self.get_base_url()}/search/simple/"
+        params = {
+            'query': query,
+            'contextLength': context_length
+        }
+        
+        def call_fn():
+            response = requests.post(url, headers=self._get_headers(), params=params, verify=self.verify_ssl, timeout=self.timeout)
+            response.raise_for_status()
+            return response.json()
+
+        return self._safe_call(call_fn)
+    
+    def append_content(self, filepath: str, content: str) -> Any:
+        url = f"{self.get_base_url()}/vault/{filepath}"
+        
+        def call_fn():
+            response = requests.post(
+                url, 
+                headers=self._get_headers() | {'Content-Type': 'text/markdown'}, 
+                data=content,
+                verify=self.verify_ssl,
+                timeout=self.timeout
+            )
+            response.raise_for_status()
+            return None
+
+        return self._safe_call(call_fn)
+    
+    def patch_content(self, filepath: str, operation: str, target_type: str, target: str, content: str) -> Any:
+        url = f"{self.get_base_url()}/vault/{filepath}"
+        
+        headers = self._get_headers() | {
+            'Content-Type': 'text/markdown',
+            'Operation': operation,
+            'Target-Type': target_type,
+            'Target': urllib.parse.quote(target)
+        }
+        
+        def call_fn():
+            response = requests.patch(url, headers=headers, data=content, verify=self.verify_ssl, timeout=self.timeout)
+            response.raise_for_status()
+            return None
+
+        return self._safe_call(call_fn)
+    
+    def search_json(self, query: dict) -> Any:
+        url = f"{self.get_base_url()}/search/"
+        
+        headers = self._get_headers() | {
+            'Content-Type': 'application/vnd.olrapi.jsonlogic+json'
+        }
+        
+        def call_fn():
+            response = requests.post(url, headers=headers, json=query, verify=self.verify_ssl, timeout=self.timeout)
+            response.raise_for_status()
+            return response.json()
+
+        return self._safe_call(call_fn)