logseq tool: use REST API instead of broken MCP endpoint
Logseq 2.0.1 has bug #12413 where POST /mcp always returns 400. Switched to the REST API at POST /api which works correctly. Requires OLLIE_LOGSEQ_TOKEN env var. Commands: listPages, getPage, searchBlocks, listTags, listProperties, upsertNodes, getCurrentGraph
This commit is contained in:
parent
c75dc6f6c7
commit
0790dc7e4e
|
|
@ -1,4 +1,5 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Logseq tool — queries and modifies Logseq DB graph via the REST API."""
|
||||
|
||||
import json
|
||||
import os
|
||||
|
|
@ -8,54 +9,72 @@ import sys
|
|||
sys.path.insert(0, os.environ.get('OLLIE_TOOLS_PATH', os.path.dirname(os.path.abspath(__file__))))
|
||||
from _lib.args import parse_args
|
||||
|
||||
RESERVED_KEYS = {"command"}
|
||||
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 to_flags(raw: dict) -> list[str]:
|
||||
"""Convert all non-reserved keys to --key=value flags."""
|
||||
flags = []
|
||||
def main():
|
||||
if not TOKEN:
|
||||
print("Error: OLLIE_LOGSEQ_TOKEN not set", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
for key, value in raw.items():
|
||||
if key in RESERVED_KEYS:
|
||||
continue
|
||||
args = parse_args()
|
||||
raw = args.raw()
|
||||
command = args.require("command")
|
||||
|
||||
if isinstance(value, str):
|
||||
try:
|
||||
parsed = json.loads(value)
|
||||
if isinstance(parsed, dict):
|
||||
for k, v in parsed.items():
|
||||
flags.append(f"--{k}={v}")
|
||||
continue
|
||||
elif isinstance(parsed, list):
|
||||
for item in parsed:
|
||||
flags.append(f"--{key}={item}")
|
||||
continue
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
pass
|
||||
flags.append(f"--{key}={value}")
|
||||
elif isinstance(value, list):
|
||||
for item in value:
|
||||
flags.append(f"--{key}={item}")
|
||||
elif isinstance(value, bool):
|
||||
if value:
|
||||
flags.append(f"--{key}")
|
||||
elif isinstance(value, dict):
|
||||
flags.append(f"--{key}={json.dumps(value)}")
|
||||
else:
|
||||
flags.append(f"--{key}={value}")
|
||||
if command not in COMMANDS:
|
||||
print(f"Error: unknown command '{command}'. Available: {', '.join(sorted(COMMANDS.keys()))}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
return flags
|
||||
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)
|
||||
|
||||
|
||||
args = parse_args()
|
||||
raw = args.raw()
|
||||
command = args.require("command")
|
||||
|
||||
cmd = ["logseq-cli", command] + to_flags(raw)
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
|
||||
if result.stdout:
|
||||
sys.stdout.write(result.stdout)
|
||||
if result.stderr:
|
||||
sys.stderr.write(result.stderr)
|
||||
sys.exit(result.returncode)
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
|
|||
Loading…
Reference in New Issue