harbor_delete_old_artifacts

Instructions

Implementation Reference

• src/harbor_registry_mcp/tools.py:674-776 (handler)

  The main handler function for harbor_delete_old_artifacts. It sorts all artifacts by push_time, keeps the N newest (keep_count), and deletes the rest. Supports dry_run mode (default true). Iterates over artifacts to delete, calling Harbor API to delete each one when dry_run=False, tracking freed size.

  def harbor_delete_old_artifacts(
      project_name: Annotated[str, Field(min_length=1, max_length=255, description="Harbor project name.")],
      repository_name: Annotated[
          str, Field(min_length=1, max_length=255, description="Repository name within the project.")
      ],
      keep_count: Annotated[int, Field(default=1, ge=0, le=1000, description="Number of newest artifacts to keep.")] = 1,
      dry_run: Annotated[
          bool, Field(default=True, description="If True (default) — only report what would be deleted, do not delete.")
      ] = True,
  ) -> DeleteOldOutput:
      """Keep the N newest artifacts in a repository, delete the rest.
  
      **DESTRUCTIVE.** ``dry_run=True`` is the default — the agent must
      explicitly set ``dry_run=False`` to actually delete. Each entry in
      ``to_delete`` carries a ``deleted`` field (``True``/``False`` after
      real run, ``None`` in dry-run).
      """
      try:
          client = get_client()
          artifacts = _list_artifacts(client, project_name, repository_name)
          sorted_arts = sorted(artifacts, key=lambda a: a.get("push_time") or "", reverse=True)
          to_keep_raw = sorted_arts[:keep_count]
          to_delete_raw = sorted_arts[keep_count:]
  
          if not to_delete_raw:
              empty: DeleteOldOutput = {
                  "project": project_name,
                  "repository": repository_name,
                  "dry_run": dry_run,
                  "keeping": [
                      {
                          "tags": [t["name"] for t in (a.get("tags") or [])],
                          "digest": a.get("digest", "")[:19],
                          "push_time": _short(a.get("push_time")),
                      }
                      for a in to_keep_raw
                  ],
                  "to_delete_count": 0,
                  "freed_size": "0 B",
                  "freed_bytes": 0,
                  "to_delete": [],
                  "hint": None,
                  "message": f"Nothing to delete — repository has {len(sorted_arts)} artifacts, keep_count={keep_count}.",
              }
              return output.ok(empty, empty["message"] or "")  # type: ignore[return-value]
  
          to_delete: list[ToDeleteArtifact] = []
          kept: list[KeptArtifact] = [
              {
                  "tags": [t["name"] for t in (a.get("tags") or [])],
                  "digest": a.get("digest", "")[:19],
                  "push_time": _short(a.get("push_time")),
              }
              for a in to_keep_raw
          ]
          total_freed = 0
  
          for art in to_delete_raw:
              digest = art.get("digest", "")
              tags = [t["name"] for t in (art.get("tags") or [])]
              size = int(art.get("size", 0) or 0)
              entry: ToDeleteArtifact = {
                  "tags": tags,
                  "digest": digest[:19],
                  "size": size_human(size),
                  "push_time": _short(art.get("push_time")),
                  "deleted": None,
              }
              if not dry_run:
                  try:
                      client.delete(
                          f"/projects/{project_name}/repositories/{encode_repo(repository_name)}/artifacts/{encode_repo(digest)}"
                      )
                      entry["deleted"] = True
                      total_freed += size
                  except Exception as e:
                      entry["deleted"] = False
                      entry["tags"].append(f"(error: {str(e)[:80]})")
              else:
                  total_freed += size
              to_delete.append(entry)
  
          result: DeleteOldOutput = {
              "project": project_name,
              "repository": repository_name,
              "dry_run": dry_run,
              "keeping": kept,
              "to_delete_count": len(to_delete),
              "freed_size": size_human(total_freed),
              "freed_bytes": total_freed,
              "to_delete": to_delete,
              "hint": "Re-run with dry_run=False to perform the deletion." if dry_run else None,
              "message": None,
          }
          prefix = "🔎 dry-run" if dry_run else "🗑 deleted"
          md = (
              f"## {prefix}: keep {keep_count} of {len(sorted_arts)} "
              f"in {project_name}/{repository_name} "
              f"({size_human(total_freed)} freed)"
          )
          return output.ok(result, md)  # type: ignore[return-value]
      except Exception as exc:
          output.fail(exc, f"deleting old artifacts in {project_name}/{repository_name}")

• src/harbor_registry_mcp/models.py:190-207 (schema)

  DeleteOldOutput TypedDict — output schema for harbor_delete_old_artifacts. Contains project, repository, dry_run, keeping (list of KeptArtifact), to_delete_count, freed_size, freed_bytes, to_delete (list of ToDeleteArtifact), hint, and message.

  class DeleteOldOutput(TypedDict):
      """Result of :func:`harbor_delete_old_artifacts`.
  
      ``dry_run=True`` means no artifacts were actually deleted — every item in
      ``to_delete`` has ``deleted=None`` and ``freed_*`` reflects hypothetical
      space reclaim. ``message`` is set when nothing matched.
      """
  
      project: str
      repository: str
      dry_run: bool
      keeping: list[KeptArtifact]
      to_delete_count: int
      freed_size: str
      freed_bytes: int
      to_delete: list[ToDeleteArtifact]
      hint: str | None
      message: str | None

• src/harbor_registry_mcp/models.py:182-187 (schema)

  ToDeleteArtifact TypedDict — schema for each artifact entry in the to_delete list. Fields: tags, digest, size, push_time, deleted (bool | None).

  class ToDeleteArtifact(TypedDict):
      tags: list[str]
      digest: str
      size: str
      push_time: str | None
      deleted: bool | None

• src/harbor_registry_mcp/models.py:176-179 (schema)

  KeptArtifact TypedDict — schema for artifacts that are kept. Fields: tags, digest, push_time.

  class KeptArtifact(TypedDict):
      tags: list[str]
      digest: str
      push_time: str | None

• src/harbor_registry_mcp/tools.py:663-672 (registration)

  Registration of the tool using @mcp.tool() decorator with name='harbor_delete_old_artifacts', annotations including destructiveHint=True, and structured_output=True.

  @mcp.tool(
      name="harbor_delete_old_artifacts",
      annotations={
          "title": "Delete Old Artifacts (keep N)",
          "readOnlyHint": False,
          "destructiveHint": True,
          "idempotentHint": False,
          "openWorldHint": True,
      },
      structured_output=True,