k8s-pilot

Instructions

Implementation Reference

• tools/cluster.py:30-48 (handler)

  The main handler function for the 'get_current_cluster' tool, decorated with @mcp.tool() to register it. It loads the kubeconfig, finds the current context, and returns the corresponding ContextInfo.
  @mcp.tool() def get_current_cluster(): """ Get the current cluster from the kubeconfig file. :return: """ config_data = get_kubeconfig() current_context = config_data.get("current-context") contexts = config_data.get("contexts", []) for ctx in contexts: if ctx["name"] == current_context: return ContextInfo( name=ctx["name"], cluster=ctx["context"].get("cluster"), user=ctx["context"].get("user"), current=True, ) return None
• server/server.py:9-9 (registration)

  Import statement in load_modules() that triggers the registration of tools from tools/cluster.py, including get_current_cluster.
  import tools.cluster # noqa: F401
• models/context.py:3-11 (schema)

  Dataclass schema/model for ContextInfo, used as the return type of the get_current_cluster tool.
  @dataclass class ContextInfo: """ Represents a Kubernetes context. """ name: str cluster: str user: str current: bool
• core/kubeconfig.py:30-44 (helper)

  Helper function get_kubeconfig() that loads and parses the kubeconfig YAML, called by the tool handler.
  def get_kubeconfig(): """ Load the kubeconfig file from the default location. The default location is usually ~/.kube/config. This function returns the parsed kubeconfig data. If the file does not exist or is not readable, it raises an exception. if you want to load a different kubeconfig file, you can set the KUBECONFIG environment variable to the path of the kubeconfig file you want to use. """ kubeconfig_path = os.path.expanduser(config.KUBE_CONFIG_DEFAULT_LOCATION) with open(kubeconfig_path, "r") as f: config_data = yaml.safe_load(f) return config_data