first tool calls working to list vault files and dir

src/mcp_knowledge_base/obsidian.py

@@ -0,0 +1,60 @@
+import requests
+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
+
+    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)
+            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)
+            response.raise_for_status()
+            
+            return response.json()['files']
+
+        return self._safe_call(call_fn)

src/mcp_knowledge_base/server.py

@@ -4,35 +4,45 @@ import logging
 from collections.abc import Sequence
 from functools import lru_cache
 from typing import Any
-
-import subprocess
+import os
 from dotenv import load_dotenv
 from mcp.server import Server
-import asyncio
 from mcp.types import (
-
+    Resource,
     Tool,
     TextContent,
     ImageContent,
     EmbeddedResource,
-    LoggingLevel
+    LoggingLevel,
 )
-from pydantic import AnyUrl
+from . import obsidian
 
 # Load environment variables
 load_dotenv()
 
-api_key = "x"
-
 # Configure logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger("mcp-knowledge-base")
 
+api_key = os.getenv("OBSIDIAN_API_KEY")
+if not api_key:
+    raise ValueError("OBSIDIAN_API_KEY environment variable required")
+
 app = Server("mcp-knowledge-base")
 
 TOOL_LIST_FILES_IN_VAULT = "list_files_in_vault"
 TOOL_LIST_FILES_IN_DIR = "list_files_in_dir"
 
+@app.list_resources()
+async def list_resources() -> list[Resource]:
+    return [
+        Resource(
+            uri="obisidian:///note/app.log",
+            name="Application Logs",
+            mimeType="text/plain"
+        )
+    ]
+
 @app.list_tools()
 async def list_tools() -> list[Tool]:
     """List available tools."""
@@ -66,20 +76,18 @@ class ToolHandler():
     def __init__(self, tool_name: str):
         self.name = tool_name
 
-    def run_tool(self, args: Any) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
+    def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
         pass
 
 class ListFilesInVaultToolHandler(ToolHandler):
     def __init__(self):
         super().__init__(TOOL_LIST_FILES_IN_VAULT)
 
-    def run_tool(self, args: Any) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
+    def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
 
-        files = [
-            "a.txt",
-            "b.txt",
-            "c/"
-        ]
+        api = obsidian.Obsidian(api_key=api_key)
+
+        files = api.list_files_in_vault()
 
         return [
             TextContent(
@@ -92,13 +100,14 @@ class ListFilesInDirToolHandler(ToolHandler):
     def __init__(self):
         super().__init__(TOOL_LIST_FILES_IN_DIR)
 
-    def run_tool(self, args: Any) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
+    def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
 
-        files = [
-            "a.txt",
-            "b.txt",
-            "c/"
-        ]
+        if "dirpath" not in args:
+            raise RuntimeError("dirpath argument missing in arguments")
+
+        api = obsidian.Obsidian(api_key=api_key)
+
+        files = api.list_files_in_dir(args["dirpath"])
 
         return [
             TextContent(
@@ -120,13 +129,15 @@ def get_tool_handler(name: str) -> ToolHandler | None:
     return tool_handlers[name]
 
 @app.call_tool()
-async def call_tool(name: str, arguments: Any) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
+async def call_tool(name: str, arguments: Any) -> Sequence[TextContent]:
     """Handle tool calls for command line run."""
     
+    if not isinstance(arguments, dict):
+        raise RuntimeError("arguments must be dictionary")
+
     add_tool_handler(ListFilesInDirToolHandler())
     add_tool_handler(ListFilesInVaultToolHandler())
 
-
     tool_handler = get_tool_handler(name)
     if not tool_handler:
         raise ValueError(f"Unknown tool: {name}")
@@ -134,8 +145,8 @@ async def call_tool(name: str, arguments: Any) -> Sequence[TextContent | ImageCo
     try:
         return tool_handler.run_tool(arguments)
     except Exception as e:
-        logger.error(f"Error: {str(e)}")
-        raise RuntimeError(f"Error: {str(e)}")
+        logger.error(str(e))
+        raise RuntimeError(f"Caught Exception. Error: {str(e)}")
 
 
 async def main():