81 lines
2.6 KiB
Python
Executable File
81 lines
2.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Logseq tool — queries and modifies Logseq DB graph via the REST API."""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from _lib.args import parse_args
|
|
|
|
LOGSEQ_URL = os.environ.get("LOGSEQ_MCP_URL", "http://127.0.0.1:12315")
|
|
TOKEN = os.environ.get("OLLIE_LOGSEQ_TOKEN", "")
|
|
|
|
# Map tool commands to REST API methods and argument shapes.
|
|
COMMANDS = {
|
|
"listPages": ("logseq.cli.listPages", lambda a: [{"expand": a.get("expand", False)}]),
|
|
"getPage": ("logseq.cli.getPageData", lambda a: [a["pageName"]]),
|
|
"searchBlocks": ("logseq.app.search", lambda a: [a["searchTerm"], {"enable-snippet?": False}]),
|
|
"listTags": ("logseq.cli.listTags", lambda a: [{"expand": a.get("expand", False)}]),
|
|
"listProperties": ("logseq.cli.listProperties", lambda a: [{"expand": a.get("expand", False)}]),
|
|
"upsertNodes": ("logseq.cli.upsertNodes", lambda a: [a["operations"], {"dry-run": a.get("dry-run", False)}]),
|
|
"getCurrentGraph":("logseq.app.getCurrentGraph", lambda a: []),
|
|
}
|
|
|
|
|
|
def main():
|
|
if not TOKEN:
|
|
print("Error: OLLIE_LOGSEQ_TOKEN not set", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
args = parse_args()
|
|
raw = args.raw()
|
|
command = args.require("command")
|
|
|
|
if command not in COMMANDS:
|
|
print(f"Error: unknown command '{command}'. Available: {', '.join(sorted(COMMANDS.keys()))}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
method, build_args = COMMANDS[command]
|
|
|
|
# Remove 'command' from raw args before passing to builder.
|
|
params = {k: v for k, v in raw.items() if k != "command"}
|
|
try:
|
|
api_args = build_args(params)
|
|
except KeyError as e:
|
|
print(f"Error: missing required argument {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
# Build the REST API request.
|
|
payload = {"method": method}
|
|
if api_args:
|
|
payload["args"] = api_args
|
|
|
|
result = subprocess.run(
|
|
["curl", "-s", "-X", "POST",
|
|
f"{LOGSEQ_URL}/api",
|
|
"-H", f"Authorization: Bearer {TOKEN}",
|
|
"-H", "Content-Type: application/json",
|
|
"-d", json.dumps(payload)],
|
|
capture_output=True, text=True, timeout=30
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
print(f"Error: curl failed: {result.stderr}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
# Output the result.
|
|
output = result.stdout.strip()
|
|
if output:
|
|
# Pretty-print JSON if possible.
|
|
try:
|
|
data = json.loads(output)
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
except json.JSONDecodeError:
|
|
print(output)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|