|
| 1 | +"""MCP root resolution via listRoots client capability. |
| 2 | +
|
| 3 | +Resolves the project root dynamically by asking the MCP client for its |
| 4 | +filesystem roots. When ``--path`` is explicitly passed on the CLI the |
| 5 | +resolver always returns that path. Otherwise it calls ``ctx.list_roots()`` |
| 6 | +on every tool invocation so it tracks workspace changes (e.g. git worktree |
| 7 | +switches) without caching stale values. |
| 8 | +""" |
| 9 | + |
| 10 | +from __future__ import annotations |
| 11 | + |
| 12 | +import logging |
| 13 | +from pathlib import Path |
| 14 | +from typing import TYPE_CHECKING |
| 15 | +from urllib.parse import unquote, urlparse |
| 16 | + |
| 17 | +if TYPE_CHECKING: |
| 18 | + from fastmcp import Context |
| 19 | + |
| 20 | +logger = logging.getLogger("deepwork.jobs.mcp") |
| 21 | + |
| 22 | + |
| 23 | +async def resolve_project_root(ctx: Context, fallback: Path) -> Path: |
| 24 | + """Ask the MCP client for its filesystem root. |
| 25 | +
|
| 26 | + Calls ``ctx.list_roots()`` and returns the first root whose URI uses the |
| 27 | + ``file`` scheme. Falls back to *fallback* when the call fails, returns |
| 28 | + no roots, or none of the roots use the ``file`` scheme. |
| 29 | + """ |
| 30 | + try: |
| 31 | + roots = await ctx.list_roots() |
| 32 | + except Exception: |
| 33 | + logger.debug("list_roots unavailable, using fallback %s", fallback) |
| 34 | + return fallback |
| 35 | + |
| 36 | + for root in roots: |
| 37 | + uri = str(root.uri) |
| 38 | + parsed = urlparse(uri) |
| 39 | + if parsed.scheme == "file": |
| 40 | + path = Path(unquote(parsed.path)).resolve() |
| 41 | + logger.debug("Resolved project root from listRoots: %s", path) |
| 42 | + return path |
| 43 | + |
| 44 | + logger.debug("No file:// root found, using fallback %s", fallback) |
| 45 | + return fallback |
| 46 | + |
| 47 | + |
| 48 | +class RootResolver: |
| 49 | + """Resolve the project root for each MCP tool call. |
| 50 | +
|
| 51 | + Parameters |
| 52 | + ---------- |
| 53 | + fallback_root: |
| 54 | + The directory to use when ``list_roots`` is unavailable (typically |
| 55 | + the process working directory or an explicit ``--path`` value). |
| 56 | + explicit: |
| 57 | + When ``True`` the *fallback_root* was explicitly provided via |
| 58 | + ``--path`` and MUST be used unconditionally. ``list_roots`` is |
| 59 | + never consulted. |
| 60 | + """ |
| 61 | + |
| 62 | + def __init__(self, fallback_root: Path, *, explicit: bool) -> None: |
| 63 | + self._fallback = fallback_root |
| 64 | + self._explicit = explicit |
| 65 | + |
| 66 | + @property |
| 67 | + def startup_root(self) -> Path: |
| 68 | + """Return the root for startup code that runs before a client connects.""" |
| 69 | + return self._fallback |
| 70 | + |
| 71 | + async def get_root(self, ctx: Context) -> Path: |
| 72 | + """Return the project root for the current tool invocation. |
| 73 | +
|
| 74 | + When ``--path`` was explicitly set, returns *fallback_root* without |
| 75 | + consulting the client. Otherwise calls ``list_roots()`` every time |
| 76 | + so workspace changes (e.g. worktree switches) are picked up |
| 77 | + immediately. |
| 78 | + """ |
| 79 | + if self._explicit: |
| 80 | + return self._fallback |
| 81 | + return await resolve_project_root(ctx, self._fallback) |
0 commit comments