Skip to content
Open
86 changes: 83 additions & 3 deletions flexus_client_kit/integrations/fi_mongo_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import os
import logging
import re
import urllib.parse
from typing import Dict, Any, Optional
from flexus_client_kit.integrations.fi_localfile import _validate_file_security

Expand All @@ -19,8 +20,8 @@
"properties": {
"op": {
"type": "string",
"enum": ["help", "list", "ls", "cat", "grep", "delete", "upload", "save"],
"description": "Operation: list/ls (list files), cat (read file), grep (search), delete, upload (from disk), save (content directly)",
"enum": ["help", "list", "ls", "cat", "grep", "delete", "upload", "save", "patch", "render_download_link"],
"description": "Operation: list/ls (list files), cat (read file), grep (search), delete, upload (from disk), save (content directly), patch (find-and-replace in a file), render_download_link (show download card to user)",
},
"args": {
"type": "object",
Expand All @@ -33,8 +34,10 @@
"pattern": {"type": ["string", "null"], "description": "Python regex pattern for grep"},
"context": {"type": ["integer", "null"], "description": "Context lines around grep matches"},
"content": {"type": ["string", "null"], "description": "Content for save op (JSON string or text)"},
"old_text": {"type": ["string", "null"], "description": "For patch: exact text to find"},
"new_text": {"type": ["string", "null"], "description": "For patch: replacement text"},
},
"required": ["path", "lines_range", "safety_valve", "pattern", "context", "content"],
"required": ["path", "lines_range", "safety_valve", "pattern", "context", "content", "old_text", "new_text"],
},
},
"required": ["op", "args"],
Expand Down Expand Up @@ -65,17 +68,42 @@
Sometimes you need to grep .json files on disk, remember that all the strings inside are escaped in that case, making
it a bit harder to match.

patch - Find and replace an exact string in a stored file.
args: path (required), old_text (required), new_text (required)
Fails if old_text is not found exactly once.

render_download_link - Show a download card to the user for an already-stored file.
The user sees a styled card with file icon, name, and download/preview button.
args: path (required)

Examples:
mongo_store(op="list", args={"path": "folder1/"})
mongo_store(op="cat", args={"path": "folder1/something_20250803.json", "lines_range": "0:40", "safety_valve": "10k"})
mongo_store(op="save", args={"path": "investigations/abc123.json", "content": "{...json...}"})
mongo_store(op="delete", args={"path": "folder1/something_20250803.json"})
mongo_store(op="grep", args={"path": "tasks.txt", "pattern": "TODO", "context": 2})
mongo_store(op="patch", args={"path": "report.html", "old_text": "<h1>Old</h1>", "new_text": "<h1>New</h1>"})
mongo_store(op="render_download_link", args={"path": "reports/monthly.pdf"})
"""

# There's also a secret op="undelete" command that can bring deleted files


_MIME_TYPES = {
".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
".webp": "image/webp", ".gif": "image/gif", ".svg": "image/svg+xml",
".pdf": "application/pdf", ".txt": "text/plain", ".csv": "text/csv",
".json": "application/json", ".html": "text/html", ".htm": "text/html",
".xml": "application/xml", ".md": "text/markdown",
".yaml": "application/yaml", ".yml": "application/yaml",
}


def _guess_mime_type(path: str) -> str:
ext = os.path.splitext(path)[1].lower()
return _MIME_TYPES.get(ext, "application/octet-stream")


async def handle_mongo_store(
rcx,
toolcall: ckit_cloudtool.FCloudtoolCall,
Expand Down Expand Up @@ -212,6 +240,58 @@ async def handle_mongo_store(
else:
return f"Error: File {path} not found in MongoDB"

elif op == "patch":
if not path:
return f"Error: path parameter required for `patch` operation\n\n{HELP}"
old_text = args.get("old_text")
new_text = args.get("new_text")
if old_text is None:
return "Error: old_text is required for patch"
if new_text is None:
return "Error: new_text is required for patch"
path_error = validate_path(path)
if path_error:
return f"Error: {path_error}"
doc = await rcx.personal_mongo.find_one({"path": path})
if not doc:
return f"Error: file not found: {path}"
if path.endswith(".json"):
json_data = doc.get("json")
if json_data is None:
return f"Error: file has no data: {path}"
content = json.dumps(json_data, indent=4)
else:
raw = doc.get("data")
if raw is None:
return f"Error: file has no data: {path}"
try:
content = bytes(raw).decode("utf-8")
except UnicodeDecodeError:
return "Error: file is not valid UTF-8 text"
count = content.count(old_text)
if count == 0:
return "Error: old_text not found in file"
if count > 1:
return f"Error: old_text found {count} times — make it more specific so it matches exactly once"
new_content = content.replace(old_text, new_text, 1)
await ckit_mongo.mongo_overwrite(rcx.personal_mongo, path, new_content.encode("utf-8"))
return f"✅ Patch applied to {path}"

elif op == "render_download_link":
if not path:
return f"Error: path parameter required for `render_download_link` operation\n\n{HELP}"
persona_id = toolcall.connected_persona_id
if not persona_id:
return "Error: `render_download_link` operation requires persona_id (connected_persona_id missing from toolcall)"
path_error = validate_path(path)
if path_error:
return f"Error: {path_error}"
display_name = os.path.basename(path)
mime = _guess_mime_type(path)
enc_path = urllib.parse.quote(path, safe="/")
enc_name = urllib.parse.quote(display_name, safe="")
return f"📎DOWNLOAD:{persona_id}:{enc_path}:{enc_name}:{mime}"

else:
return HELP

Expand Down