kill-process-mcp

Instructions

Implementation Reference

• kill_process_mcp.py:164-225 (handler)

  The core handler function for the 'process_list' tool. Decorated with @mcp.tool(), it handles input parameters, validates them, samples CPU usage, collects and serializes processes using psutil, applies filters (name, user, status, CPU, memory, system), sorts by CPU or memory usage, applies limit, and returns the list of process dictionaries.

  @mcp.tool()
  async def process_list(sort_by: Literal["cpu", "memory"] = "cpu", duration: int | str = 2, limit: int | str | None = None, name_filter: str | None = None, user_filter: str | None = None, status_filter: Literal["running", "sleeping", "stopped", "zombie"] | None = None, min_cpu: float | str | None = None, min_memory: int | str | None = None, include_system: bool = False, sort_asc: bool = False, ctx: Context | None = None,) -> List[Dict[str, Any]]:
      """List running processes sorted by CPU or memory with optional name, user, status, CPU/memory thresholds, system-process filtering, sort order and limit."""
  
      if ctx:
          await ctx.info(
              f"process_list called sort_by={sort_by} duration={duration} (type={type(duration)}) "
              f"limit={limit} (type={type(limit)}) name_filter={name_filter} user_filter={user_filter} "
              f"status_filter={status_filter} min_cpu={min_cpu} (type={type(min_cpu)}) "
              f"min_memory={min_memory} (type={type(min_memory)}) include_system={include_system} "
              f"sort_asc={sort_asc}"
          )
  
      if sort_by not in {"cpu", "memory"}:
          raise ValueError(f"sort_by must be 'cpu' or 'memory', got: {sort_by} (type: {type(sort_by)})")
  
      if limit is not None:
          limit = _to_int(limit, "limit")
          if limit < 0:
              raise ValueError(f"limit must be non-negative, got: {limit}")
  
      duration = _to_int(duration, "duration")
      if duration < 0:
          raise ValueError(f"duration must be non-negative, got: {duration}")
  
      if min_cpu is not None:
          min_cpu = _to_float(min_cpu, "min_cpu")
          if min_cpu < 0:
              raise ValueError(f"min_cpu must be non-negative, got: {min_cpu}")
  
      if min_memory is not None:
          min_memory = _to_int(min_memory, "min_memory")
          if min_memory < 0:
              raise ValueError(f"min_memory must be non-negative, got: {min_memory}")
  
      _snapshot_cpu()
  
      await asyncio.sleep(max(0.5, duration if sort_by == "cpu" else duration))
  
      procs = _collect_processes()
      serialised = [_serialize(p) for p in procs]
  
      if not include_system:
          serialised = [p for p in serialised if p["username"] not in SYSTEM_USERS]
      if name_filter is not None:
          serialised = [p for p in serialised if name_filter.lower() in p["name"].lower()]
      if user_filter is not None:
          serialised = [p for p in serialised if user_filter.lower() in p["username"].lower()]
      if status_filter is not None:
          serialised = [p for p in serialised if p["status"] == status_filter]
      if min_cpu is not None:
          serialised = [p for p in serialised if p["cpu_percent"] >= min_cpu]
      if min_memory is not None:
          serialised = [p for p in serialised if p["rss"] >= min_memory]
  
      key = "cpu_percent" if sort_by == "cpu" else "rss"
      result = sorted(serialised, key=lambda p: p[key], reverse=not sort_asc)
  
      if limit is not None:
          result = result[:limit]
  
      return result

• kill_process_mcp.py:165-166 (schema)

  Input schema defined by the function parameters with type annotations (Literal, int|str, etc.) and comprehensive docstring describing usage. Output is List[Dict[str, Any]] containing process info (pid, name, username, status, cpu_percent, rss).

  async def process_list(sort_by: Literal["cpu", "memory"] = "cpu", duration: int | str = 2, limit: int | str | None = None, name_filter: str | None = None, user_filter: str | None = None, status_filter: Literal["running", "sleeping", "stopped", "zombie"] | None = None, min_cpu: float | str | None = None, min_memory: int | str | None = None, include_system: bool = False, sort_asc: bool = False, ctx: Context | None = None,) -> List[Dict[str, Any]]:
      """List running processes sorted by CPU or memory with optional name, user, status, CPU/memory thresholds, system-process filtering, sort order and limit."""

• kill_process_mcp.py:164-164 (registration)

  The @mcp.tool() decorator on process_list registers it as an MCP tool with FastMCP instance.

  @mcp.tool()

• kill_process_mcp.py:106-137 (helper)

  Helper function to serialize psutil.Process objects into JSON-safe dicts with pid, name, username, status, cpu_percent, and rss (resident set size or physical footprint on macOS). Used in process_list.

  def _serialize(proc: psutil.Process) -> Dict[str, Any]:
      try:
          try:
              mem = _phys_footprint(proc.pid)
          except Exception:
              mem = 0
          if not mem:
              try:
                  mi = proc.memory_full_info()
                  mem = getattr(mi, "uss", mi.rss)
              except Exception:
                  try:
                      mem = proc.memory_info().rss
                  except Exception:
                      mem = 0
  
          if sys.platform == "win32":
              try:
                  exe_name = os.path.basename(proc.exe())
                  name = exe_name if exe_name else proc.name()
              except (psutil.Error, OSError, FileNotFoundError):
                  name = proc.name()
          else:
              try:
                  name = proc.name()
              except psutil.Error:
                  name = "<unknown>"
  
          cpu = proc.cpu_percent(None)
          return {"pid": proc.pid, "name": name, "username": proc.username(), "status": proc.status(), "cpu_percent": cpu, "rss": mem}
      except psutil.Error:
          return {"pid": proc.pid, "name": "<terminated>", "username": "<unknown>", "status": "<terminated>", "cpu_percent": 0.0, "rss": 0}

• kill_process_mcp.py:95-102 (helper)

  Helper to safely collect a list of current psutil.Process instances, skipping errors.

  def _collect_processes() -> List[psutil.Process]:
      procs: List[psutil.Process] = []
      for proc in psutil.process_iter(attrs=["pid", "username"]):
          try:
              procs.append(proc)
          except psutil.Error:
              continue
      return procs