Expose transfers as tools and check them in doctor

CHANGELOG.md

@@ -5,6 +5,26 @@ uses Semantic Versioning for public releases.
 
 ## Unreleased
 ### Added
+- `transfer_ls`, `transfer_put`, `transfer_get` and `transfer_status` expose
+  data movement as tools, in the deferred pool; the core surface stays at 33.
+  Three verbs rather than one `transfer(op=...)` dispatcher, because a model
+  picks better from schemas that name their own arguments and an upload is not
+  the same authorization decision as a download. `transfer_status` is the
+  fourth because the other two return as soon as Globus accepts the task, and
+  without it the returned `task_id` names something nothing can read back. They
+  register only when some endpoint declares a `globus_transfer` block: a tool
+  whose only possible answer is "not configured" is still a tool the model can
+  call, so an install that moves no files shows no sign of them, and the
+  namespace-coverage check excuses exactly those four names rather than
+  whatever happens to be unregistered. A refused path comes back as a result
+  with its reason rather than a traceback, and nothing is submitted before the
+  paths are checked, so a rejected request costs no network and leaves no
+  half-made task. `validate_hpc_setup` (behind `doctor`) gains a transfer
+  check: it passes when nothing is configured, since transfers are opt-in the
+  way HPC is, and fails on a configured transfer that cannot work -- no write
+  root, no SDK, no consent, or a write root the collection will not list. The
+  write root is readable as well as writable, which the doctor probe found the
+  hard way: somewhere you may put a file is somewhere you may look at one.
 - Files can now move between this machine and an HPC collection. Compute has
   always run there and nothing could get a file there or back: a mesh had to be
   staged by hand before a remote tool could see it, and a subset or export a

config.yaml.example

@@ -39,6 +39,32 @@ hpc:
       path_prefixes:
         - "/lcrc/"
         - "/home/<user>/"
+      # Optional: where this endpoint's files live, in Globus Transfer's terms.
+      # Omit the block entirely and the endpoint runs code but moves no files --
+      # the transfer tools are not registered at all. A Compute endpoint UUID
+      # says nothing about which collection serves that filesystem, so the
+      # collection is named explicitly rather than inferred.
+      globus_transfer:
+        # The Globus collection that serves this cluster's filesystem.
+        remote_collection_id: "00000000-0000-0000-0000-000000000000"
+        # This machine's collection (Globus Connect Personal). Without it
+        # nothing here is one end of a transfer.
+        local_collection_id: "00000000-0000-0000-0000-000000000000"
+        # Uploads must resolve inside this. It is a boundary, not a default
+        # directory: a path that escapes it is refused, not relocated.
+        remote_write_root: "/lcrc/group/e3sm/<user>"
+        # Optional: widens where downloads may read from without widening
+        # where uploads may write. The write root stays readable either way.
+        remote_read_root: "/lcrc/group/e3sm"
+        # Optional: a collection that exposes a subtree calls that subtree "/".
+        # Longest match wins; a path under none of these is passed to Globus
+        # unchanged, because a wrong translation names a real file nobody
+        # asked for while an unrecognized one is merely rejected.
+        collection_roots:
+          - "/lcrc/group/e3sm"
+        # Optional: bounds where downloads may land on this machine, checked
+        # after symlinks are resolved.
+        local_root: "/Users/<user>/Downloads"
 
   # ---- Legacy single-endpoint form (still supported) ----
   # Keep ``hpc.endpoints`` empty and set this if you only ever submit to one

src/uxarray_mcp/registry.py

@@ -138,6 +138,35 @@
     "hpc": ("check_remote_yac",),
 }
 
+# Deferred too, but only when some endpoint declares a ``globus_transfer``
+# block. An install that moves no files should show no sign of these: a tool
+# whose only possible answer is "not configured" is still a tool the model can
+# call, and calling it is the wrong lesson to have taught it.
+_CONDITIONAL_TOOLS: dict[str, tuple[str, ...]] = {
+    "transfer": (
+        "transfer_ls",
+        "transfer_put",
+        "transfer_get",
+        "transfer_status",
+    ),
+}
+
+_CONDITIONAL_NAMES: frozenset[str] = frozenset(
+    name for names in _CONDITIONAL_TOOLS.values() for name in names
+)
+
+
+def _transfers_are_configured() -> bool:
+    """Whether any endpoint says where its files live.
+
+    Wrapped rather than imported at module scope so building a registry never
+    depends on config being readable, and so a test can decide the answer
+    without writing a config file.
+    """
+    from uxarray_mcp.tools.transfer_tools import transfers_are_configured
+
+    return transfers_are_configured()
+
 
 # ---------------------------------------------------------------------------
 # Prompt-as-tool helpers (formerly @mcp.prompt() decorators)
@@ -638,6 +667,10 @@ def _apply_output_schema(tool: object, raw_name: str) -> None:
 
 _SEARCH_HINTS: dict[str, str] = {
     "check_remote_yac": "yac native remap conservative interpolation worker library build smoke test hpc",
+    "transfer_ls": "list remote directory globus collection files hpc browse",
+    "transfer_put": "upload stage copy file to hpc cluster globus transfer send",
+    "transfer_get": "download fetch retrieve file from hpc cluster globus transfer",
+    "transfer_status": "transfer task progress bytes globus poll",
     "calculate_curl": "vorticity rotation circulation wind curl cross product compute vector field zeta",
     "calculate_divergence": "compression expansion source sink wind divergence",
     "calculate_gradient": "spatial derivative slope field gradient",
@@ -760,6 +793,18 @@ def build_registry(
                 search_hint=_SEARCH_HINTS.get(raw, ""),
             )
             registered.add(raw)
+        if _transfers_are_configured():
+            for ns, raw in _flatten(_CONDITIONAL_TOOLS):
+                func = getattr(_tools_mod, raw)
+                registry.register(func, namespace=ns)
+                qualified = f"{ns}{sep}{raw}"
+                _apply_tags(registry, qualified, raw, func)
+                registry.update_tool_metadata(
+                    qualified,
+                    defer=True,
+                    search_hint=_SEARCH_HINTS.get(raw, ""),
+                )
+                registered.add(raw)
         registry.enable_tool_discovery()
 
     # ``enable_tool_discovery`` registers ``discover_tools`` itself, so it
@@ -795,7 +840,11 @@ def _verify_coverage(registered: set[str], profile: Profile) -> None:
                 f"Bridge tried to register non-public tools: {sorted(bogus)}"
             )
         return
-    missing = public - registered
+    # The conditional tools are public so they can be imported and tested, but
+    # absent from an unconfigured registry on purpose. Excusing them here is
+    # narrower than excusing whatever happens not to be registered: anything
+    # else missing is still the loud failure this check exists to be.
+    missing = public - registered - _CONDITIONAL_NAMES
     if missing:
         raise RuntimeError(
             f"Namespace plan out of date — {len(missing)} public tools "

src/uxarray_mcp/remote/transfer.py

@@ -276,19 +276,41 @@ def _write_root(self) -> str:
             )
         return root
 
-    def _read_root(self) -> str:
-        """Reads fall back to the write root, never to the whole filesystem."""
-        return self.profile.remote_read_root or self._write_root()
+    def _read_roots(self) -> tuple[str, ...]:
+        """Where reads may come from, in the order a relative path resolves.
+
+        The write root is always readable: somewhere you may put a file is
+        somewhere you may look at one, and a config that could write to
+        `/scratch` but not list it would fail on the first download of
+        something it had just uploaded. With no read root configured, reads
+        fall back to the write root -- never to the whole filesystem.
+        """
+        write_root = self.profile.remote_write_root
+        read_root = self.profile.remote_read_root
+        if not read_root:
+            return (self._write_root(),)
+        if write_root and write_root != read_root:
+            return (read_root, write_root)
+        return (read_root,)
 
     def remote_write_path(self, path: str) -> str:
         return to_collection_path(
             join_under(self._write_root(), path), self.profile.collection_roots
         )
 
     def remote_read_path(self, path: str) -> str:
-        return to_collection_path(
-            join_under(self._read_root(), path), self.profile.collection_roots
-        )
+        roots = self._read_roots()
+        last: PathOutsideRoot | None = None
+        for root in roots:
+            try:
+                resolved = join_under(root, path)
+            except PathOutsideRoot as exc:
+                last = exc
+                continue
+            return to_collection_path(resolved, self.profile.collection_roots)
+        raise PathOutsideRoot(
+            f"{path!r} is outside every readable root ({', '.join(roots)})."
+        ) from last
 
     def _local_collection_id(self) -> str:
         if not self.profile.local_collection_id:

src/uxarray_mcp/tools/__init__.py

@@ -68,6 +68,12 @@
     resume_workflow,
     run_workflow,
 )
+from .transfer_tools import (
+    transfer_get,
+    transfer_ls,
+    transfer_put,
+    transfer_status,
+)
 from .vector_calc import (
     calculate_azimuthal_mean,
     calculate_curl,
@@ -135,4 +141,8 @@
     "manage_session",
     "get_status",
     "get_result",
+    "transfer_ls",
+    "transfer_put",
+    "transfer_get",
+    "transfer_status",
 ]

src/uxarray_mcp/tools/execution_control.py

@@ -62,6 +62,110 @@ def _make_check(
     return result
 
 
+def _transfer_check(
+    base_config: Any, endpoint: str | None, run_probe: bool
+) -> Dict[str, Any]:
+    """Report whether this endpoint can move data, not just run code.
+
+    Passing when nothing is configured is deliberate: transfers are opt-in the
+    way HPC itself is, and a doctor that goes red for a feature the user never
+    asked for teaches people to ignore it. What is worth failing on is a
+    configured transfer that cannot work -- SDK absent, no consent, or a write
+    root the collection will not show.
+
+    The reachability probe rides on ``run_remote_probe`` because it is a real
+    network call, and it lists the write root rather than transferring
+    anything: a listing proves the collection, the consent and the path all
+    line up, and moves no bytes.
+    """
+    try:
+        profile = base_config.resolve_endpoint(endpoint=endpoint)
+    except Exception:
+        profile = None
+    transfer_profile = getattr(profile, "globus_transfer", None)
+    if transfer_profile is None:
+        return _make_check(
+            "transfer",
+            True,
+            "No globus_transfer block configured; this endpoint moves no files.",
+            details={"configured": False},
+            guidance=(
+                "Add hpc.endpoints.<name>.globus_transfer with "
+                "remote_collection_id and remote_write_root to enable "
+                "transfer_put / transfer_get."
+            ),
+        )
+
+    details: Dict[str, Any] = {
+        "configured": True,
+        "endpoint_name": getattr(profile, "name", None),
+        "remote_collection_id": transfer_profile.remote_collection_id,
+        "local_collection_id": transfer_profile.local_collection_id,
+        "remote_write_root": transfer_profile.remote_write_root,
+        "remote_read_root": transfer_profile.remote_read_root,
+        "collection_roots": list(transfer_profile.collection_roots),
+    }
+
+    if not transfer_profile.remote_write_root:
+        return _make_check(
+            "transfer",
+            False,
+            "globus_transfer is configured without a remote_write_root, so "
+            "uploads have nowhere they are allowed to land.",
+            details=details,
+            guidance="Set remote_write_root on this endpoint's globus_transfer block.",
+        )
+
+    from uxarray_mcp.remote.transfer import TransferService
+
+    service = TransferService(transfer_profile)
+    try:
+        service.client  # noqa: B018 -- builds the client, checks login state
+    except Exception as exc:
+        return _make_check(
+            "transfer",
+            False,
+            "Globus Transfer is configured but no client could be built.",
+            details={**details, **_exception_details(exc)},
+            guidance=(
+                "Install the transfer extra (`uv sync --extra transfer`) and "
+                "complete the Globus login in a terminal; an MCP server cannot "
+                "open a browser consent flow."
+            ),
+        )
+
+    if not run_probe:
+        return _make_check(
+            "transfer",
+            True,
+            "Globus Transfer client is authenticated; collection reachability "
+            "not probed.",
+            details=details,
+        )
+
+    try:
+        entries = service.ls(transfer_profile.remote_write_root)
+    except Exception as exc:
+        return _make_check(
+            "transfer",
+            False,
+            f"Write root {transfer_profile.remote_write_root!r} could not be "
+            f"listed on the collection.",
+            details={**details, **_exception_details(exc)},
+            guidance=(
+                "Check that remote_collection_id serves this filesystem, that "
+                "collection_roots translate the path the way the collection "
+                "names it, and that the write root exists."
+            ),
+        )
+    return _make_check(
+        "transfer",
+        True,
+        f"Collection reachable; write root lists {len(entries)} entries.",
+        details={**details, "entry_count": len(entries)},
+    )
+
+
 def _guidance_for_error(message: str) -> str | None:
     """Return targeted next-step guidance for common HPC setup failures."""
     lowered = message.lower()
@@ -662,6 +766,8 @@ def validate_hpc_setup(
                 )
             )
 
+    checks.append(_transfer_check(base_config, endpoint, run_remote_probe))
+
     passed = all(check["passed"] for check in checks)
     result = {
         "passed": passed,