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