From dfc0eb319b0f6283dc4581f303db7e3fb743a98d Mon Sep 17 00:00:00 2001 From: Markus Pfundstein Date: Fri, 29 Nov 2024 14:31:05 +0100 Subject: [PATCH] modifying files --- README.md | 8 +- src/mcp_knowledge_base/obsidian.py | 46 ++++++++ src/mcp_knowledge_base/server.py | 3 + src/mcp_knowledge_base/tools.py | 162 ++++++++++++++++++++++++++++- 4 files changed, 216 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 83b6bc9..f66c285 100644 --- a/README.md +++ b/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`
diff --git a/src/mcp_knowledge_base/obsidian.py b/src/mcp_knowledge_base/obsidian.py index f5e6e2e..6b6f094 100644 --- a/src/mcp_knowledge_base/obsidian.py +++ b/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) \ No newline at end of file diff --git a/src/mcp_knowledge_base/server.py b/src/mcp_knowledge_base/server.py index 4d980f0..c3e0936 100644 --- a/src/mcp_knowledge_base/server.py +++ b/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]: diff --git a/src/mcp_knowledge_base/tools.py b/src/mcp_knowledge_base/tools.py index 989a67a..b923250 100644 --- a/src/mcp_knowledge_base/tools.py +++ b/src/mcp_knowledge_base/tools.py @@ -125,4 +125,164 @@ class GetFileContentsToolHandler(ToolHandler): type="text", text=json.dumps(content, indent=2) ) - ] \ No newline at end of file + ] + +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']}" + ) + ] \ No newline at end of file