server.py

#!/usr/bin/env python3
"""
projectfs - a per-project filesystem tool for Claude Desktop.

One local MCP server that lets you attach specific folders to specific
"projects" and then read/write files only inside the folders attached to
the project you name. Solves the fact that Claude Desktop's local MCP
servers are otherwise global with no per-project scoping.

Two ways to run it:

  * No arguments  -> runs as an MCP stdio server (this is what Claude
    Desktop does via claude_desktop_config.json).

  * With arguments -> a small CLI you use in Terminal to manage the
    project -> folder mappings:

        python3 server.py init
        python3 server.py add <project>
        python3 server.py link <project> <folder>
        python3 server.py unlink <project> <folder>
        python3 server.py rm <project>
        python3 server.py list
        python3 server.py admin-token

Config path, in order of precedence:
  1. $PROJECTFS_CONFIG
  2. config.json next to this file (if it exists)
  3. ~/.projectfs/config.json

  {
    "projects": {
      "my-notes": {
        "token": "pf_...",             # identifies the project to file ops
        "folders": ["/Users/you/Documents/Notes"]
      }
    }
  }

Security model (read this):

  * File operations (read/write/list/append/delete/move) require the
    calling chat to pass the project's `project_token`. Nothing can be
    read or written outside the folders attached to that project.
  * Attaching / detaching folders is CLI-only (`link` / `unlink` in a
    terminal). There is deliberately no MCP tool for it: that would let any
    chat that could read an admin token re-scope a project to an arbitrary
    directory. `list_projects` is the only mapping-related tool, and it
    shows project names and folders but never the tokens.
  * Paths are fully resolved (symlinks included) and must stay inside an
    attached folder; `..` escapes are rejected. File operations re-open the
    resolved path component-by-component with O_NOFOLLOW, so a symlink
    swapped in after the check cannot redirect a write outside the folder.
  * Directory delete/move is refused. Read is capped at 5 MiB, writes at
    10 MiB. Non-UTF-8 files are reported, not dumped.

Pure standard library. Works on the system Python (/usr/bin/python3).
"""

import sys
import os
import io
import json
import errno
import secrets
import stat
import traceback

HERE = os.path.dirname(os.path.abspath(__file__))

def _config_path():
    # 1. explicit override
    env = os.environ.get("PROJECTFS_CONFIG")
    if env:
        return os.path.expanduser(env)
    # 2. back-compat: a config.json sitting next to this script
    beside = os.path.join(HERE, "config.json")
    if os.path.exists(beside):
        return beside
    # 3. default: ~/.projectfs/config.json  (used by the .mcpb bundle install)
    return os.path.expanduser("~/.projectfs/config.json")

CONFIG_PATH = _config_path()

READ_CAP = 5 * 1024 * 1024
WRITE_CAP = 10 * 1024 * 1024

SERVER_NAME = "projectfs"
SERVER_VERSION = "1.0.0"
DEFAULT_PROTOCOL = "2025-06-18"


# --------------------------------------------------------------------------
# config
# --------------------------------------------------------------------------

class ConfigError(Exception):
    pass


def load_config():
    if not os.path.exists(CONFIG_PATH):
        return {"admin_token": "", "projects": {}}
    try:
        with open(CONFIG_PATH, "r", encoding="utf-8") as f:
            data = json.load(f)
    except Exception as e:
        raise ConfigError("cannot read %s: %s" % (CONFIG_PATH, e))
    if not isinstance(data, dict):
        raise ConfigError("config root must be an object")
    data.setdefault("admin_token", "")
    data.setdefault("projects", {})
    if not isinstance(data["projects"], dict):
        raise ConfigError("config 'projects' must be an object")
    # normalise
    for name, entry in list(data["projects"].items()):
        if not isinstance(entry, dict):
            raise ConfigError("project %r must be an object" % name)
        entry.setdefault("token", "")
        entry.setdefault("folders", [])
        if not isinstance(entry["folders"], list):
            raise ConfigError("project %r folders must be a list" % name)
    return data


def save_config(data):
    parent = os.path.dirname(CONFIG_PATH)
    if parent and not os.path.isdir(parent):
        os.makedirs(parent, exist_ok=True)
    tmp = CONFIG_PATH + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2)
        f.write("\n")
    os.replace(tmp, CONFIG_PATH)
    try:
        os.chmod(CONFIG_PATH, 0o600)
    except OSError:
        pass


def new_token(prefix):
    return prefix + "_" + secrets.token_urlsafe(18)


def find_project_by_token(cfg, token):
    if not token:
        return None, None
    for name, entry in cfg["projects"].items():
        if entry.get("token") and secrets.compare_digest(str(entry["token"]), str(token)):
            return name, entry
    return None, None


# --------------------------------------------------------------------------
# path handling
# --------------------------------------------------------------------------

def real(p):
    return os.path.realpath(os.path.abspath(os.path.expanduser(p)))


def allowed_roots(entry):
    roots = []
    for f in entry.get("folders", []):
        roots.append(real(f))
    return roots


def resolve_within(entry, path):
    """Resolve `path` and ensure it is inside one of the project's folders.

    Relative paths resolve against the project's first folder.
    Returns the resolved absolute path or raises ValueError.
    """
    roots = allowed_roots(entry)
    if not roots:
        raise ValueError("this project has no folders attached yet")

    if os.path.isabs(path) or path.startswith("~"):
        target = real(path)
    else:
        target = real(os.path.join(roots[0], path))

    for root in roots:
        if target == root or target.startswith(root + os.sep):
            return target
    raise ValueError(
        "path is outside this project's folders: %s\nattached folders:\n  %s"
        % (target, "\n  ".join(roots))
    )


# --------------------------------------------------------------------------
# race-free traversal
#
# resolve_within() resolves symlinks with os.path.realpath and confirms the
# result is inside an attached folder, but every tool then performed a *second*
# filesystem call on that path string. A local process that can write inside the
# folder could swap a path component for a symlink in the gap between the two,
# redirecting a write/delete/move outside the sandbox (a TOCTOU).
#
# The helpers below close that gap: they re-open the already-resolved canonical
# path one component at a time, relative to a file descriptor, with O_NOFOLLOW.
# realpath already collapsed every legitimate (in-sandbox) symlink, so a symlink
# encountered during this walk means someone is racing us - the open fails
# instead of escaping. All subsequent I/O then goes through the resulting
# dir_fd + leaf name, never a path string.
# --------------------------------------------------------------------------

_ODIR = getattr(os, "O_DIRECTORY", 0)
_ONOFOLLOW = getattr(os, "O_NOFOLLOW", 0)

# dir_fd + O_NOFOLLOW are POSIX; on a platform without them (e.g. Windows) fall
# back to the path-based behaviour rather than break.
_SAFE_TRAVERSAL = bool(_ONOFOLLOW) and os.open in getattr(os, "supports_dir_fd", set())


def _split_within_root(entry, target):
    """Split an already containment-checked `target` into (root, [components])."""
    for root in allowed_roots(entry):
        if target == root:
            return root, []
        if target.startswith(root + os.sep):
            return root, target[len(root) + 1:].split(os.sep)
    raise ValueError("path is outside this project's folders: %s" % target)


def _descend(root, components, create_parents=False):
    """Walk `components[:-1]` down from `root` without following any symlink.

    Returns (dir_fd, leaf). The caller owns dir_fd and must os.close() it.
    `leaf` is components[-1], or os.curdir when `components` is empty (target is
    the root itself). Raises ValueError if a component is a symlink (or missing,
    unless create_parents).
    """
    dir_fd = os.open(root, os.O_RDONLY | _ODIR | _ONOFOLLOW)
    try:
        for comp in components[:-1]:
            if comp in ("", os.curdir, os.pardir):
                raise ValueError("unsafe path component: %r" % comp)
            try:
                nxt = os.open(comp, os.O_RDONLY | _ODIR | _ONOFOLLOW, dir_fd=dir_fd)
            except FileNotFoundError:
                if not create_parents:
                    raise ValueError(
                        "no such directory: %s" % os.path.join(root, *components[:-1])
                    )
                os.mkdir(comp, dir_fd=dir_fd)
                nxt = os.open(comp, os.O_RDONLY | _ODIR | _ONOFOLLOW, dir_fd=dir_fd)
            except NotADirectoryError:
                raise ValueError("not a directory: %s" % comp)
            except OSError as e:
                if e.errno == errno.ELOOP:
                    raise ValueError(
                        "refusing to follow a symlink inside the project: %s/%s"
                        % (root, "/".join(components))
                    )
                raise
            os.close(dir_fd)
            dir_fd = nxt
        leaf = components[-1] if components else os.curdir
        if leaf in ("", os.pardir):
            raise ValueError("unsafe path leaf: %r" % leaf)
        return dir_fd, leaf
    except BaseException:
        os.close(dir_fd)
        raise


# --------------------------------------------------------------------------
# tool implementations  (return plain strings; raise ValueError for user errors)
# --------------------------------------------------------------------------

def tool_list_projects(cfg, args):
    projects = cfg["projects"]
    if not projects:
        return "No projects configured yet. Use the CLI:\n  python3 ~/projectfs/server.py add <project>"
    lines = []
    for name in sorted(projects):
        entry = projects[name]
        folders = entry.get("folders", [])
        lines.append("%s" % name)
        if folders:
            for fol in folders:
                exists = "" if os.path.isdir(real(fol)) else "  (MISSING)"
                lines.append("    - %s%s" % (fol, exists))
        else:
            lines.append("    (no folders attached)")
    return "\n".join(lines)


# Attaching / detaching folders is intentionally CLI-only (see the `link` /
# `unlink` subcommands). Exposing it as an MCP tool - even behind an admin token -
# means any chat that can read that token could re-point a project at an
# arbitrary directory, which defeats the scoping this server exists to provide.


def _project_from_args(cfg, args):
    token = args.get("project_token", "")
    name, entry = find_project_by_token(cfg, token)
    if entry is None:
        raise ValueError("unknown or missing project_token")
    return name, entry


def _open_leaf(dir_fd, leaf, flags, mode=0o666):
    """os.open(leaf, ..., dir_fd=dir_fd) with friendlier errors. O_NOFOLLOW is
    added so a leaf that got swapped for a symlink after the check fails here."""
    try:
        return os.open(leaf, flags | _ONOFOLLOW, mode, dir_fd=dir_fd)
    except OSError as e:
        if e.errno == errno.ELOOP:
            raise ValueError("refusing to follow a symlink at the target path")
        raise


def tool_list_dir(cfg, args):
    name, entry = _project_from_args(cfg, args)
    path = args.get("path", ".")
    target = resolve_within(entry, path)

    if not _SAFE_TRAVERSAL:
        if not os.path.isdir(target):
            raise ValueError("not a directory: %s" % target)
        names = sorted(os.listdir(target))
        stat_at = lambda nm: os.lstat(os.path.join(target, nm))
    else:
        root, comps = _split_within_root(entry, target)
        dir_fd, leaf = _descend(root, comps)
        try:
            if leaf == os.curdir:
                list_fd = os.dup(dir_fd)
            else:
                try:
                    list_fd = os.open(leaf, os.O_RDONLY | _ODIR | _ONOFOLLOW, dir_fd=dir_fd)
                except FileNotFoundError:
                    raise ValueError("no such path: %s" % target)
                except NotADirectoryError:
                    raise ValueError("not a directory: %s" % target)
                except OSError as e:
                    if e.errno == errno.ELOOP:
                        raise ValueError("refusing to follow a symlink: %s" % target)
                    raise
        finally:
            os.close(dir_fd)
        try:
            names = sorted(os.listdir(list_fd))
            entries = [(nm, os.stat(nm, dir_fd=list_fd, follow_symlinks=False)) for nm in names]
        finally:
            os.close(list_fd)
        rows = []
        for nm, st in entries:
            if stat.S_ISDIR(st.st_mode):
                rows.append("%s/" % nm)
            elif stat.S_ISLNK(st.st_mode):
                rows.append("%s@  (symlink, not followed)" % nm)
            else:
                rows.append("%s  (%d bytes)" % (nm, st.st_size))
        header = "%s  [project: %s]" % (target, name)
        return header + "\n" + ("\n".join(rows) if rows else "(empty)")

    # fallback path
    rows = []
    for nm in names:
        try:
            st = stat_at(nm)
        except OSError:
            rows.append("%s  (stat failed)" % nm)
            continue
        if stat.S_ISDIR(st.st_mode):
            rows.append("%s/" % nm)
        elif stat.S_ISLNK(st.st_mode):
            rows.append("%s@  (symlink, not followed)" % nm)
        else:
            rows.append("%s  (%d bytes)" % (nm, st.st_size))
    header = "%s  [project: %s]" % (target, name)
    return header + "\n" + ("\n".join(rows) if rows else "(empty)")


def tool_read_file(cfg, args):
    name, entry = _project_from_args(cfg, args)
    path = args.get("path")
    if not path:
        raise ValueError("path is required")
    target = resolve_within(entry, path)

    if not _SAFE_TRAVERSAL:
        if not os.path.isfile(target) or os.path.islink(target):
            raise ValueError("no such file: %s" % target)
        fd = os.open(target, os.O_RDONLY)
    else:
        root, comps = _split_within_root(entry, target)
        dir_fd, leaf = _descend(root, comps)
        try:
            try:
                fd = _open_leaf(dir_fd, leaf, os.O_RDONLY)
            except FileNotFoundError:
                raise ValueError("no such file: %s" % target)
        finally:
            os.close(dir_fd)

    with os.fdopen(fd, "rb") as f:
        st = os.fstat(f.fileno())
        if not stat.S_ISREG(st.st_mode):
            raise ValueError("not a regular file: %s" % target)
        if st.st_size > READ_CAP:
            raise ValueError("file is %d bytes, over the %d byte read limit" % (st.st_size, READ_CAP))
        raw = f.read(READ_CAP + 1)
    if len(raw) > READ_CAP:
        raise ValueError("file is over the %d byte read limit" % READ_CAP)
    try:
        return raw.decode("utf-8")
    except UnicodeDecodeError:
        return "[binary or non-UTF-8 file, %d bytes, not shown: %s]" % (len(raw), target)


def _write_within(entry, path, data, append):
    blob = data.encode("utf-8")
    if len(blob) > WRITE_CAP:
        raise ValueError("content exceeds the %d byte write limit" % WRITE_CAP)
    target = resolve_within(entry, path)
    flags = os.O_WRONLY | os.O_CREAT | (os.O_APPEND if append else os.O_TRUNC)

    if not _SAFE_TRAVERSAL:
        if os.path.isdir(target):
            raise ValueError("path is a directory: %s" % target)
        parent = os.path.dirname(target)
        resolve_within(entry, parent)  # parent must also be inside a root
        if not os.path.isdir(parent):
            os.makedirs(parent, exist_ok=True)
        fd = os.open(target, flags | _ONOFOLLOW, 0o666)
    else:
        root, comps = _split_within_root(entry, target)
        if not comps:
            raise ValueError("refusing to write to the project folder itself: %s" % target)
        dir_fd, leaf = _descend(root, comps, create_parents=True)
        try:
            try:
                fd = _open_leaf(dir_fd, leaf, flags)
            except IsADirectoryError:
                raise ValueError("path is a directory: %s" % target)
        finally:
            os.close(dir_fd)

    with os.fdopen(fd, "ab" if append else "wb") as f:
        f.write(blob)
    return target


def tool_write_file(cfg, args):
    name, entry = _project_from_args(cfg, args)
    path = args.get("path")
    if not path:
        raise ValueError("path is required")
    content = args.get("content", "")
    target = _write_within(entry, path, content, append=False)
    return "wrote %d bytes to %s" % (len(content.encode("utf-8")), target)


def tool_append_file(cfg, args):
    name, entry = _project_from_args(cfg, args)
    path = args.get("path")
    if not path:
        raise ValueError("path is required")
    content = args.get("content", "")
    target = _write_within(entry, path, content, append=True)
    return "appended %d bytes to %s" % (len(content.encode("utf-8")), target)


def tool_delete_file(cfg, args):
    name, entry = _project_from_args(cfg, args)
    path = args.get("path")
    if not path:
        raise ValueError("path is required")
    target = resolve_within(entry, path)

    if not _SAFE_TRAVERSAL:
        if os.path.isdir(target) and not os.path.islink(target):
            raise ValueError("refusing to delete a directory: %s" % target)
        if not os.path.lexists(target):
            raise ValueError("no such file: %s" % target)
        os.remove(target)
        return "deleted %s" % target

    root, comps = _split_within_root(entry, target)
    if not comps:
        raise ValueError("refusing to delete the project folder: %s" % target)
    dir_fd, leaf = _descend(root, comps)
    try:
        try:
            st = os.stat(leaf, dir_fd=dir_fd, follow_symlinks=False)
        except FileNotFoundError:
            raise ValueError("no such file: %s" % target)
        if stat.S_ISDIR(st.st_mode):
            raise ValueError("refusing to delete a directory: %s" % target)
        if stat.S_ISLNK(st.st_mode):
            raise ValueError("refusing to delete a symlink: %s" % target)
        os.unlink(leaf, dir_fd=dir_fd)
    finally:
        os.close(dir_fd)
    return "deleted %s" % target


def tool_move_file(cfg, args):
    name, entry = _project_from_args(cfg, args)
    source = args.get("source")
    dest = args.get("dest")
    if not source or not dest:
        raise ValueError("source and dest are required")
    s = resolve_within(entry, source)
    d = resolve_within(entry, dest)

    if not _SAFE_TRAVERSAL:
        if os.path.isdir(s) and not os.path.islink(s):
            raise ValueError("refusing to move a directory: %s" % s)
        if not os.path.lexists(s):
            raise ValueError("no such file: %s" % s)
        dparent = os.path.dirname(d)
        resolve_within(entry, dparent)
        if not os.path.isdir(dparent):
            os.makedirs(dparent, exist_ok=True)
        os.rename(s, d)
        return "moved %s -> %s" % (s, d)

    s_root, s_comps = _split_within_root(entry, s)
    d_root, d_comps = _split_within_root(entry, d)
    if not s_comps:
        raise ValueError("refusing to move the project folder: %s" % s)
    if not d_comps:
        raise ValueError("refusing to overwrite the project folder: %s" % d)
    s_dir_fd, s_leaf = _descend(s_root, s_comps)
    try:
        try:
            st = os.stat(s_leaf, dir_fd=s_dir_fd, follow_symlinks=False)
        except FileNotFoundError:
            raise ValueError("no such file: %s" % s)
        if stat.S_ISDIR(st.st_mode):
            raise ValueError("refusing to move a directory: %s" % s)
        d_dir_fd, d_leaf = _descend(d_root, d_comps, create_parents=True)
        try:
            try:
                os.rename(s_leaf, d_leaf, src_dir_fd=s_dir_fd, dst_dir_fd=d_dir_fd)
            except OSError as e:
                if e.errno in (errno.EISDIR, errno.ENOTEMPTY, errno.EEXIST):
                    raise ValueError("destination is an existing directory: %s" % d)
                raise
        finally:
            os.close(d_dir_fd)
    finally:
        os.close(s_dir_fd)
    return "moved %s -> %s" % (s, d)


TOOLS = [
    {
        "name": "list_projects",
        "description": (
            "List the projectfs projects and the local folders attached to each. "
            "Does not reveal any tokens."
        ),
        "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
        "impl": tool_list_projects,
    },
    # No link_folder / unlink_folder tool on purpose - folder mapping is CLI-only
    # so a chat can never re-scope a project (see note by _project_from_args).
    {
        "name": "list_dir",
        "description": (
            "List a directory inside the folders attached to a project. "
            "path is optional and defaults to the project's first folder."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "project_token": {"type": "string", "description": "the project's token"},
                "path": {"type": "string", "description": "absolute path inside an attached folder, or relative to the first attached folder"},
            },
            "required": ["project_token"],
            "additionalProperties": False,
        },
        "impl": tool_list_dir,
    },
    {
        "name": "read_file",
        "description": "Read a UTF-8 text file inside a project's attached folders.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "project_token": {"type": "string"},
                "path": {"type": "string"},
            },
            "required": ["project_token", "path"],
            "additionalProperties": False,
        },
        "impl": tool_read_file,
    },
    {
        "name": "write_file",
        "description": (
            "Create or overwrite a text file inside a project's attached folders. "
            "Parent directories are created as needed (still inside an attached folder)."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "project_token": {"type": "string"},
                "path": {"type": "string"},
                "content": {"type": "string"},
            },
            "required": ["project_token", "path", "content"],
            "additionalProperties": False,
        },
        "impl": tool_write_file,
    },
    {
        "name": "append_file",
        "description": "Append text to a file inside a project's attached folders (creates it if absent). Good for running logs.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "project_token": {"type": "string"},
                "path": {"type": "string"},
                "content": {"type": "string"},
            },
            "required": ["project_token", "path", "content"],
            "additionalProperties": False,
        },
        "impl": tool_append_file,
    },
    {
        "name": "delete_file",
        "description": "Delete a single file inside a project's attached folders. Directories are refused.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "project_token": {"type": "string"},
                "path": {"type": "string"},
            },
            "required": ["project_token", "path"],
            "additionalProperties": False,
        },
        "impl": tool_delete_file,
    },
    {
        "name": "move_file",
        "description": "Move or rename a file. Both source and dest must be inside the project's attached folders.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "project_token": {"type": "string"},
                "source": {"type": "string"},
                "dest": {"type": "string"},
            },
            "required": ["project_token", "source", "dest"],
            "additionalProperties": False,
        },
        "impl": tool_move_file,
    },
]

TOOLS_BY_NAME = {t["name"]: t for t in TOOLS}


# --------------------------------------------------------------------------
# MCP stdio server  (newline-delimited JSON-RPC 2.0)
# --------------------------------------------------------------------------

def log(*a):
    sys.stderr.write("[projectfs] " + " ".join(str(x) for x in a) + "\n")
    sys.stderr.flush()


def rpc_result(id_, result):
    return {"jsonrpc": "2.0", "id": id_, "result": result}


def rpc_error(id_, code, message):
    return {"jsonrpc": "2.0", "id": id_, "error": {"code": code, "message": message}}


def handle_message(msg):
    method = msg.get("method")
    id_ = msg.get("id")
    params = msg.get("params") or {}
    is_notification = "id" not in msg

    if method == "initialize":
        proto = params.get("protocolVersion") or DEFAULT_PROTOCOL
        return rpc_result(id_, {
            "protocolVersion": proto,
            "capabilities": {"tools": {"listChanged": False}},
            "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION},
        })

    if method in ("notifications/initialized", "notifications/cancelled", "initialized"):
        return None

    if method == "ping":
        return rpc_result(id_, {})

    if method == "tools/list":
        listed = []
        for t in TOOLS:
            listed.append({
                "name": t["name"],
                "description": t["description"],
                "inputSchema": t["inputSchema"],
            })
        return rpc_result(id_, {"tools": listed})

    if method == "resources/list":
        return rpc_result(id_, {"resources": []})

    if method == "prompts/list":
        return rpc_result(id_, {"prompts": []})

    if method == "tools/call":
        name = params.get("name")
        args = params.get("arguments") or {}
        tool = TOOLS_BY_NAME.get(name)
        if tool is None:
            return rpc_result(id_, {
                "content": [{"type": "text", "text": "unknown tool: %s" % name}],
                "isError": True,
            })
        try:
            cfg = load_config()
        except ConfigError as e:
            return rpc_result(id_, {
                "content": [{"type": "text", "text": "config error: %s" % e}],
                "isError": True,
            })
        try:
            out = tool["impl"](cfg, args)
            return rpc_result(id_, {"content": [{"type": "text", "text": out}], "isError": False})
        except ValueError as e:
            return rpc_result(id_, {"content": [{"type": "text", "text": str(e)}], "isError": True})
        except OSError as e:  # permission, missing path, etc. - report cleanly
            hint = ""
            if getattr(e, "errno", None) in (1, 13):  # EPERM / EACCES
                hint = (
                    "\n\nThis looks like a macOS privacy (TCC) block. If the folder is "
                    "under ~/Desktop, ~/Documents, or ~/Downloads, either move it "
                    "somewhere else (e.g. ~/) or grant Claude Full Disk Access in "
                    "System Settings > Privacy & Security, then restart Claude."
                )
            log("tool %s OSError: %s" % (name, e))
            return rpc_result(id_, {
                "content": [{"type": "text", "text": "filesystem error: %s%s" % (e, hint)}],
                "isError": True,
            })
        except Exception as e:  # noqa
            log("tool %s crashed: %s" % (name, traceback.format_exc()))
            return rpc_result(id_, {
                "content": [{"type": "text", "text": "internal error: %s" % e}],
                "isError": True,
            })

    if is_notification:
        return None
    return rpc_error(id_, -32601, "method not found: %s" % method)


def serve():
    log("starting; config at", CONFIG_PATH)
    try:
        cfg = load_config()
        log("projects:", ", ".join(sorted(cfg["projects"])) or "(none)")
    except ConfigError as e:
        log("WARNING config not loadable:", e)

    stdin = io.TextIOWrapper(sys.stdin.buffer, encoding="utf-8")
    stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", newline="\n")
    for line in stdin:
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except Exception as e:
            log("bad json:", e)
            continue
        try:
            resp = handle_message(msg)
        except Exception as e:  # noqa
            log("handler crashed:", traceback.format_exc())
            resp = rpc_error(msg.get("id"), -32603, "internal error: %s" % e)
        if resp is not None:
            stdout.write(json.dumps(resp) + "\n")
            stdout.flush()
    log("stdin closed; exiting")


# --------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------

def cli(argv):
    cmd = argv[0]

    if cmd in ("-h", "--help", "help"):
        print(__doc__)
        return 0

    if cmd == "init":
        try:
            cfg = load_config()
        except ConfigError:
            cfg = {"admin_token": "", "projects": {}}
        if not cfg.get("admin_token"):
            cfg["admin_token"] = new_token("adm")
        save_config(cfg)
        print("config: %s" % CONFIG_PATH)
        print("admin_token: %s" % cfg["admin_token"])
        print("\nKeep the admin token private. You only need it to link/unlink")
        print("folders from inside a chat; the CLI here does not require it.")
        return 0

    cfg = load_config()

    if cmd == "list":
        if not cfg["projects"]:
            print("(no projects)  create one:  python3 %s add <project>" % os.path.basename(__file__))
            return 0
        for name in sorted(cfg["projects"]):
            e = cfg["projects"][name]
            print("== %s ==" % name)
            print("   project_token: %s" % e.get("token", ""))
            if e["folders"]:
                for fol in e["folders"]:
                    flag = "" if os.path.isdir(real(fol)) else "   <-- MISSING"
                    print("   folder: %s%s" % (fol, flag))
            else:
                print("   (no folders)")
            print("   --- paste into this project's Claude instructions: ---")
            print("   For all file operations use the projectfs tools with")
            print("   project_token \"%s\". Save every document you produce as a" % e.get("token", ""))
            print("   file in this project's folder via write_file.")
            print()
        return 0

    if cmd == "admin-token":
        print(cfg.get("admin_token") or "(none - run: init)")
        return 0

    if cmd == "add":
        if len(argv) < 2:
            print("usage: add <project>")
            return 2
        name = argv[1].strip()
        if name in cfg["projects"]:
            print("already exists: %s (token %s)" % (name, cfg["projects"][name]["token"]))
            return 0
        cfg["projects"][name] = {"token": new_token("pf"), "folders": []}
        save_config(cfg)
        print("created %s" % name)
        print("project_token: %s" % cfg["projects"][name]["token"])
        print("now attach a folder:  python3 %s link %s <folder>" % (os.path.basename(__file__), name))
        return 0

    if cmd == "rm":
        if len(argv) < 2:
            print("usage: rm <project>")
            return 2
        name = argv[1].strip()
        if name not in cfg["projects"]:
            print("no such project: %s" % name)
            return 1
        del cfg["projects"][name]
        save_config(cfg)
        print("removed %s" % name)
        return 0

    if cmd == "link":
        if len(argv) < 3:
            print("usage: link <project> <folder>")
            return 2
        name = argv[1].strip()
        folder = real(argv[2])
        if not os.path.isdir(folder):
            print("not a directory: %s" % folder)
            return 1
        e = cfg["projects"].get(name)
        if e is None:
            e = {"token": new_token("pf"), "folders": []}
            cfg["projects"][name] = e
            print("created project %s" % name)
        if folder in [real(x) for x in e["folders"]]:
            print("already attached")
            return 0
        e["folders"].append(folder)
        save_config(cfg)
        print("attached %s -> %s" % (name, folder))
        print("project_token: %s" % e["token"])
        return 0

    if cmd == "unlink":
        if len(argv) < 3:
            print("usage: unlink <project> <folder>")
            return 2
        name = argv[1].strip()
        folder = real(argv[2])
        e = cfg["projects"].get(name)
        if e is None:
            print("no such project: %s" % name)
            return 1
        n0 = len(e["folders"])
        e["folders"] = [x for x in e["folders"] if real(x) != folder]
        if len(e["folders"]) == n0:
            print("folder was not attached: %s" % folder)
            return 1
        save_config(cfg)
        print("detached %s from %s" % (folder, name))
        return 0

    print("unknown command: %s (try --help)" % cmd)
    return 2


def main():
    if len(sys.argv) > 1:
        sys.exit(cli(sys.argv[1:]))
    serve()


if __name__ == "__main__":
    main()