modifying files

README.md

@@ -7,9 +7,13 @@ Example MCP server to interact with Obsidian.
 ### Tools
 
 The server implements multiple tools to interact with Obsidian:
+
 - list_files_in_vault: Lists all files and directories in the root directory of your Obsidian vault
 - list_files_in_dir: Lists all files and directories in a specific Obsidian directory
 - get_file_contents: Return the content of a single file in your vault.
+- search: Search for documents matching a specified text query across all files in the vault
+- patch_content: Insert content into an existing note relative to a heading, block reference, or frontmatter field.
+- append_content: Append content to a new or existing file in the vault.
 
 ### Example prompts
 
@@ -18,11 +22,10 @@ Its good to first instruct Claude to use Obsidian. Then it will always call the
 - List all files in my vault
 - List all files in the XYZ directory
 - Get the contents of the last architecture call note and summarize them
+- Search for all files where Azure CosmosDb is mentioned and quickly explain to me the context in which it is mentioned
 
 ## Configuration
 
-
-
 ### Environment Variables
 
 Create a `.env` file in the root directory with the following required variable:
@@ -46,6 +49,7 @@ Install and enable it in the settings and copy the api key.
 #### Claude Desktop
 
 On MacOS: `~/Library/Application\ Support/Claude/claude_desktop_config.json`
+
 On Windows: `%APPDATA%/Claude/claude_desktop_config.json`
 
 <details>

src/mcp_knowledge_base/obsidian.py

@@ -68,4 +68,50 @@ class Obsidian():
             
             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)
+            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
+            )
+            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': target
+        }
+        
+        def call_fn():
+            response = requests.patch(url, headers=headers, data=content, verify=self.verify_ssl)
+            response.raise_for_status()
+            return None
+
         return self._safe_call(call_fn)

src/mcp_knowledge_base/server.py

@@ -47,6 +47,9 @@ def get_tool_handler(name: str) -> tools.ToolHandler | None:
 add_tool_handler(tools.ListFilesInDirToolHandler())
 add_tool_handler(tools.ListFilesInVaultToolHandler())
 add_tool_handler(tools.GetFileContentsToolHandler())
+add_tool_handler(tools.SearchToolHandler())
+add_tool_handler(tools.PatchContentToolHandler())
+add_tool_handler(tools.AppendContentToolHandler())
 
 #@app.list_resources()
 #async def list_resources() -> list[Resource]:

src/mcp_knowledge_base/tools.py

@@ -125,4 +125,164 @@ class GetFileContentsToolHandler(ToolHandler):
                 type="text",
                 text=json.dumps(content, indent=2)
             )
-        ]
+        ]
+    
+class SearchToolHandler(ToolHandler):
+    def __init__(self):
+        super().__init__("search")
+
+    def get_tool_description(self):
+        return Tool(
+            name=self.name,
+            description="Search for documents matching a specified text query across all files in the vault",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "query": {
+                        "type": "string",
+                        "description": "Text to search for in the vault"
+                    },
+                    "context_length": {
+                        "type": "integer",
+                        "description": "How much context to return around the matching string (default: 100)",
+                        "default": 100
+                    }
+                },
+                "required": ["query"]
+            }
+        )
+
+    def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
+        if "query" not in args:
+            raise RuntimeError("query argument missing in arguments")
+
+        context_length = args.get("context_length", 100)
+        
+        api = obsidian.Obsidian(api_key=api_key)
+        results = api.search(args["query"], context_length)
+        
+        formatted_results = []
+        for result in results:
+            formatted_matches = []
+            for match in result.get('matches', []):
+                context = match.get('context', '')
+                match_pos = match.get('match', {})
+                start = match_pos.get('start', 0)
+                end = match_pos.get('end', 0)
+                
+                formatted_matches.append({
+                    'context': context,
+                    'match_position': {'start': start, 'end': end}
+                })
+                
+            formatted_results.append({
+                'filename': result.get('filename', ''),
+                'score': result.get('score', 0),
+                'matches': formatted_matches
+            })
+
+        return [
+            TextContent(
+                type="text",
+                text=json.dumps(formatted_results, indent=2)
+            )
+        ]
+    
+class AppendContentToolHandler(ToolHandler):
+   def __init__(self):
+       super().__init__("append_content")
+
+   def get_tool_description(self):
+       return Tool(
+           name=self.name,
+           description="Append content to a new or existing file in the vault.",
+           inputSchema={
+               "type": "object",
+               "properties": {
+                   "filepath": {
+                       "type": "string",
+                       "description": "Path to the file (relative to vault root)",
+                       "format": "path"
+                   },
+                   "content": {
+                       "type": "string",
+                       "description": "Content to append to the file"
+                   }
+               },
+               "required": ["filepath", "content"]
+           }
+       )
+
+   def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
+       if "filepath" not in args or "content" not in args:
+           raise RuntimeError("filepath and content arguments required")
+
+       api = obsidian.Obsidian(api_key=api_key)
+       api.append_content(args["filepath"], args["content"])
+
+       return [
+           TextContent(
+               type="text",
+               text=f"Successfully appended content to {args['filepath']}"
+           )
+       ]
+   
+class PatchContentToolHandler(ToolHandler):
+   def __init__(self):
+       super().__init__("patch_content")
+
+   def get_tool_description(self):
+       return Tool(
+           name=self.name,
+           description="Insert content into an existing note relative to a heading, block reference, or frontmatter field.",
+           inputSchema={
+               "type": "object",
+               "properties": {
+                   "filepath": {
+                       "type": "string",
+                       "description": "Path to the file (relative to vault root)",
+                       "format": "path"
+                   },
+                   "operation": {
+                       "type": "string",
+                       "description": "Operation to perform (append, prepend, or replace)",
+                       "enum": ["append", "prepend", "replace"]
+                   },
+                   "target_type": {
+                       "type": "string",
+                       "description": "Type of target to patch",
+                       "enum": ["heading", "block", "frontmatter"]
+                   },
+                   "target": {
+                       "type": "string", 
+                       "description": "Target identifier (heading path, block reference, or frontmatter field)"
+                   },
+                   "content": {
+                       "type": "string",
+                       "description": "Content to insert"
+                   }
+               },
+               "required": ["filepath", "operation", "target_type", "target", "content"]
+           }
+       )
+
+   def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
+       required = ["filepath", "operation", "target_type", "target", "content"]
+       if not all(key in args for key in required):
+           raise RuntimeError(f"Missing required arguments: {', '.join(required)}")
+
+       api = obsidian.Obsidian(api_key=api_key)
+       api.patch_content(
+           args["filepath"],
+           args["operation"],
+           args["target_type"],
+           args["target"],
+           args["content"]
+       )
+
+       return [
+           TextContent(
+               type="text",
+               text=f"Successfully patched content in {args['filepath']}"
+           )
+       ]