Skip to content

refactor(client): move driver introspection into jumpstarter.client for CLI/MCP parity #1042

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
kirkbrauer wants to merge 8 commits into
base: main
Choose a base branch
from
Open

refactor(client): move driver introspection into jumpstarter.client for CLI/MCP parity #1042

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
7a862ed
refactor(client): move driver introspection into jumpstarter.client f…
kirkbrauer Aug 29, 2026
0dfc27c
fix(client): drive lease introspection from outside the event loop
kirkbrauer Aug 30, 2026
fb7b060
feat(client): describe command parameters well enough to prompt for them
kirkbrauer Aug 30, 2026
5d9e792
refactor(client): address review on the introspection move
kirkbrauer Aug 31, 2026
1dd009a
test: cover introspection review fixes across MCP and client APIs
kirkbrauer Sep 5, 2026
080009e
fix(client): retain declared help inputs and valid root call examples
kirkbrauer Sep 5, 2026
b51f524
fix(client): preserve structured mapping defaults in introspection
kirkbrauer Sep 5, 2026
cf391f2
fix(client): reject colliding JSON mapping keys in defaults
kirkbrauer Sep 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
230 changes: 25 additions & 205 deletions python/packages/jumpstarter-mcp/jumpstarter_mcp/introspect.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,205 +1,25 @@
"""Introspection utilities for Click CLI trees and driver object trees."""

from __future__ import annotations

import inspect
import logging
from typing import Any

import click
import click.core

logger = logging.getLogger(__name__)


def walk_click_tree(cmd: click.core.BaseCommand, path: list[str] | None = None) -> dict[str, Any]: # ty: ignore[unresolved-attribute]
"""Recursively walk a Click command tree and return structured JSON.

Returns command names, help text, parameters (with types and defaults),
and nested subcommands.
"""
path = path or []
result: dict[str, Any] = {
"name": cmd.name,
"help": cmd.help,
"params": [
{
"name": p.name,
"type": str(p.type),
"help": getattr(p, "help", None),
"required": getattr(p, "required", False),
"default": p.default if p.default is not None else None,
}
for p in cmd.params
if not getattr(p, "hidden", False) and p.name != "help"
],
}
if isinstance(cmd, click.Group):
result["subcommands"] = {
name: walk_click_tree(sub, path + [name])
for name, sub in cmd.commands.items()
}
return result


# Methods on the base DriverClient that should be excluded from introspection
_BASE_METHODS = {
"call",
"streamingcall",
"stream",
"stream_async",
"log_stream",
"log_stream_async",
"open_stream",
"close",
"reset",
"cli",
"call_async",
"streamingcall_async",
"report",
"report_async",
"get_status_async",
"end_session_async",
"status_monitor_async",
}

# Methods inherited from base classes that add noise to driver listings
_INTERNAL_METHODS = {
"check_exporter_status",
"resource_async",
"wait_for_hook_complete_monitored",
"wait_for_hook_status",
"wait_for_lease_ready",
"wait_for_lease_ready_monitored",
}


def _get_public_method_names(obj: Any) -> list[str]:
"""Return public method names for a driver instance, excluding base internals.

Uses getmembers_static to avoid triggering property descriptors, which can
make gRPC calls that fail when invoked from the event loop thread.
"""
names = []
try:
members = inspect.getmembers_static(obj)
except Exception:
logger.debug("inspect.getmembers_static failed for %s", type(obj).__name__, exc_info=True)
return names
for name, value in members:
if name.startswith("_") or name in _BASE_METHODS or name in _INTERNAL_METHODS:
continue
if isinstance(value, property):
continue
if not callable(value):
continue
try:
inspect.signature(value)
except (ValueError, TypeError, AttributeError):
continue
names.append(name)
return names


def list_drivers(client: Any, _keys: list[str] | None = None) -> list[dict[str, Any]]:
"""Flatten the driver client tree into a list with dot-separated paths.

Returns path (Python access path like ``client.power``), driver_path
(children-key list like ``["power"]`` for use with get_driver_methods),
class name, description, and method names for each driver.
"""
keys = _keys or []
cls = type(client)
results = [
{
"path": f"client.{'.'.join(keys)}" if keys else "client",
"driver_path": keys,
"class": f"{cls.__module__}.{cls.__qualname__}",
"description": getattr(client, "description", None),
"methods": _get_public_method_names(client),
}
]
children = getattr(client, "children", {})
for name, child in children.items():
results.extend(list_drivers(child, keys + [name]))
return results


def _inspect_method(name: str, method: Any, driver_path: list[str]) -> dict[str, Any] | None:
"""Build a method descriptor dict, or return None if the method can't be inspected."""
try:
sig = inspect.signature(method)
except (ValueError, TypeError, AttributeError):
return None

is_streaming = False
try:
source = inspect.getsource(method)
is_streaming = "streamingcall" in source
except (OSError, TypeError):
pass

params = [
{
"name": pname,
"annotation": str(p.annotation) if p.annotation != inspect.Parameter.empty else None,
"default": str(p.default) if p.default != inspect.Parameter.empty else None,
}
for pname, p in sig.parameters.items()
if pname != "self"
]

attr_path = ".".join(driver_path)
call_args = ", ".join(f"{p['name']}=..." for p in params)
method_call = f"client.{attr_path}.{name}({call_args})"

return {
"name": name,
"signature": str(sig),
"docstring": inspect.getdoc(method),
"parameters": params,
"return_type": str(sig.return_annotation) if sig.return_annotation != inspect.Signature.empty else None,
"is_streaming": is_streaming,
"call_example": (
"from jumpstarter.utils.env import env\n\n"
"with env() as client:\n"
f" {method_call}"
),
}


def get_driver_methods(client: Any, driver_path: list[str]) -> dict[str, Any]:
"""Inspect a specific driver client at the given path in the children tree.

Uses Python inspect to return detailed method information for all public
methods defined on the concrete driver class.
"""
target = client
for key in driver_path:
children = getattr(target, "children", {})
if key not in children:
raise KeyError(f"Driver path component '{key}' not found. Available: {list(children.keys())}")
target = children[key]

cls = type(target)
try:
members = inspect.getmembers_static(target)
except Exception:
logger.debug("inspect.getmembers_static failed for %s", cls.__name__, exc_info=True)
members = []

methods = []
for name, value in members:
if name.startswith("_") or name in _BASE_METHODS or name in _INTERNAL_METHODS:
continue
if isinstance(value, property) or not callable(value):
continue
info = _inspect_method(name, value, driver_path)
if info is not None:
methods.append(info)

return {
"class": f"{cls.__module__}.{cls.__qualname__}",
"driver_path": driver_path,
"methods": methods,
}
"""Backward-compatible re-exports of the driver introspection utilities.

The implementation lives in jumpstarter.client.introspect so it can be shared
between the CLI and the MCP server.
"""

from jumpstarter.client.introspect import (
_BASE_METHODS,
_INTERNAL_METHODS,
_get_public_method_names,
_inspect_method,
get_driver_methods,
list_drivers,
walk_click_tree,
)

__all__ = [
"_BASE_METHODS",
"_INTERNAL_METHODS",
"_get_public_method_names",
"_inspect_method",
"get_driver_methods",
"list_drivers",
"walk_click_tree",
]
140 changes: 9 additions & 131 deletions python/packages/jumpstarter-mcp/jumpstarter_mcp/server_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,16 +10,9 @@
from datetime import datetime
from unittest.mock import AsyncMock, MagicMock, patch

import click
import pytest

from jumpstarter_mcp.connections import Connection, ConnectionManager
from jumpstarter_mcp.introspect import (
_get_public_method_names,
get_driver_methods,
list_drivers,
walk_click_tree,
)
from jumpstarter_mcp.server import (
TOKEN_REFRESH_THRESHOLD_SECONDS,
_ensure_fresh_token,
Expand Down Expand Up @@ -112,135 +105,20 @@ class FakeLease:


# ---------------------------------------------------------------------------
# walk_click_tree
# Introspection re-exports
# ---------------------------------------------------------------------------


class TestWalkClickTree:
def test_simple_command(self):
@click.command("hello")
@click.option("--name", help="Your name")
def hello(name):
"""Say hello."""

result = walk_click_tree(hello)
assert result["name"] == "hello"
assert result["help"] == "Say hello."
assert len(result["params"]) == 1
assert result["params"][0]["name"] == "name"
assert result["params"][0]["help"] == "Your name"
assert "subcommands" not in result

def test_group_with_subcommands(self):
@click.group("root")
def root():
"""Root group."""

@root.command("sub1")
def sub1():
"""First sub."""

@root.command("sub2")
@click.option("--count", type=int, default=5)
def sub2(count):
"""Second sub."""

result = walk_click_tree(root)
assert result["name"] == "root"
assert "subcommands" in result
assert "sub1" in result["subcommands"]
assert "sub2" in result["subcommands"]
assert result["subcommands"]["sub2"]["params"][0]["name"] == "count"
assert result["subcommands"]["sub2"]["params"][0]["default"] == 5

def test_hidden_params_excluded(self):
@click.command("cmd")
@click.option("--visible", help="shown")
@click.option("--secret", hidden=True)
def cmd(visible, secret):
pass

result = walk_click_tree(cmd)
names = [p["name"] for p in result["params"]]
assert "visible" in names
assert "secret" not in names


# Introspection tests
# ---------------------------------------------------------------------------
class TestIntrospectReexports:
def test_reexports_core_implementation(self):
import jumpstarter_mcp.introspect as shim

import jumpstarter.client.introspect as core

class TestGetPublicMethodNames:
def test_filters_private_and_base_methods(self):
names = _get_public_method_names(FakePowerClient())
assert "on" in names
assert "off" in names
assert "cycle" in names
assert "_private" not in names
assert "call" not in names
assert "check_exporter_status" not in names


class TestListDrivers:
def test_flat_tree(self):
client = FakeCompositeClient()
result = list_drivers(client)
paths = [d["path"] for d in result]
assert "client" in paths
assert "client.power" in paths
assert "client.serial" in paths

def test_driver_path_field(self):
client = FakeCompositeClient()
result = list_drivers(client)
root = next(d for d in result if d["path"] == "client")
assert root["driver_path"] == []
power = next(d for d in result if d["path"] == "client.power")
assert power["driver_path"] == ["power"]

def test_methods_populated(self):
client = FakeCompositeClient()
result = list_drivers(client)
power = next(d for d in result if d["path"] == "client.power")
assert "on" in power["methods"]
assert "off" in power["methods"]


class TestGetDriverMethods:
def test_returns_method_details(self):
client = FakeCompositeClient()
result = get_driver_methods(client, ["power"])
assert result["driver_path"] == ["power"]
method_names = [m["name"] for m in result["methods"]]
assert "on" in method_names
assert "off" in method_names
assert "cycle" in method_names

def test_call_example_uses_dot_notation(self):
client = FakeCompositeClient()
result = get_driver_methods(client, ["power"])
cycle = next(m for m in result["methods"] if m["name"] == "cycle")
assert "client.power.cycle(" in cycle["call_example"]
assert 'children["power"]' not in cycle["call_example"]

def test_invalid_path_raises(self):
client = FakeCompositeClient()
with pytest.raises(KeyError, match="nonexistent"):
get_driver_methods(client, ["nonexistent"])

def test_docstrings_captured(self):
client = FakeCompositeClient()
result = get_driver_methods(client, ["power"])
on_method = next(m for m in result["methods"] if m["name"] == "on")
assert on_method["docstring"] == "Power on the device."

def test_parameters_captured(self):
client = FakeCompositeClient()
result = get_driver_methods(client, ["power"])
cycle = next(m for m in result["methods"] if m["name"] == "cycle")
assert len(cycle["parameters"]) == 1
assert cycle["parameters"][0]["name"] == "wait"
assert cycle["parameters"][0]["default"] == "2"
assert shim.walk_click_tree is core.walk_click_tree
assert shim.list_drivers is core.list_drivers
assert shim.get_driver_methods is core.get_driver_methods
assert shim._get_public_method_names is core._get_public_method_names


# ---------------------------------------------------------------------------
Expand Down
Loading
Loading