"""Milestone and checkpoint tools."""
from typing import Any, Optional
from mcp.server.fastmcp import FastMCP
from .. import client
from ..types import ClinkApiError, Milestone, is_hil_required
def _format_milestone(m: Milestone, include_checkpoints: bool = True) -> str:
"""Format a milestone for display."""
output = f"Title: {m.title}\n"
output += f"ID: {m.milestone_id}\n"
if m.project_id:
output += f"Project ID: {m.project_id}\n"
output += f"Status: {m.status}\n"
output += f"Progress: {m.progress}\n"
if include_checkpoints and m.checkpoints:
output += "\nCheckpoints:\n"
for cp in m.checkpoints:
annotations = []
if cp.requires_proposal_id:
annotations.append("requires consensus")
if cp.depends_on:
annotations.append(f"depends on: {', '.join(str(d) for d in cp.depends_on)}")
annotation_str = f" [{'; '.join(annotations)}]" if annotations else ""
output += f" {cp.order}. {cp.title}{annotation_str} - {cp.status}\n"
if cp.git_branch_url:
output += f" Branch: {cp.git_branch_url}\n"
if cp.git_pr_url:
output += f" PR: {cp.git_pr_url}\n"
if cp.git_commit_url:
output += f" Commit: {cp.git_commit_url}\n"
return output
def register(mcp: FastMCP) -> None:
"""Register milestone tools with the MCP server."""
@mcp.tool()
async def create_milestone(
group: str,
title: str,
checkpoints: list[dict[str, Any]],
description: Optional[str] = None,
project_id: Optional[str] = None,
) -> str:
"""Create a milestone with checkpoints to track multi-step collaborative tasks. Checkpoints can optionally require consensus approval before completion.
Args:
group: The group slug (e.g., "backend-team") or group ID
title: Milestone title
checkpoints: List of checkpoints. Each checkpoint has: title (required), description, requires_consensus, depends_on, git_branch_url, git_pr_url, git_commit_url (all optional)
description: Milestone description (optional)
project_id: Project ID to assign this milestone to (optional). If not provided, uses the group's default "General" project.
"""
try:
if not group:
groups = await client.list_groups()
if groups:
group_list = "\n".join(f'* {g.slug} - "{g.name}"' for g in groups)
return f"Please provide a group slug. Your groups:\n\n{group_list}"
return "Please provide a group slug. You are not a member of any groups."
if not title:
return "Please provide a milestone title."
if not checkpoints or len(checkpoints) == 0:
return "Please provide at least one checkpoint."
if len(checkpoints) > 50:
return "Maximum 50 checkpoints per milestone."
# Resolve slug or ID to group ID
group_id = await client.resolve_group(group)
milestone = await client.create_milestone(
group_id,
title=title,
checkpoints=checkpoints,
description=description,
project_id=project_id,
)
output = "Milestone created successfully!\n\n"
output += _format_milestone(milestone)
return output
except ClinkApiError as e:
return f"Error: {e}"
@mcp.tool()
async def list_milestones(
group: str,
status: Optional[str] = None,
limit: Optional[int] = None,
) -> str:
"""List milestones for a Clink group. Shows progress and status of each milestone.
Args:
group: The group slug (e.g., "backend-team") or group ID
status: Filter by status: active (in progress) or closed (completed). Default: all.
limit: Maximum milestones to return (default: 20)
"""
try:
if not group:
groups = await client.list_groups()
if groups:
group_list = "\n".join(f'* {g.slug} - "{g.name}"' for g in groups)
return f"Please provide a group slug. Your groups:\n\n{group_list}"
return "Please provide a group slug."
group_id = await client.resolve_group(group)
milestones = await client.list_milestones(
group_id,
status=status, # type: ignore
limit=limit,
)
if not milestones:
return f"No milestones found in group '{group}'."
output = f"Milestones in {group} ({len(milestones)}):\n\n"
for m in milestones:
output += f"* {m.title}\n"
output += f" ID: {m.milestone_id}\n"
output += f" Status: {m.status} | Progress: {m.progress}\n\n"
return output.strip()
except ClinkApiError as e:
return f"Error: {e}"
@mcp.tool()
async def get_milestone(milestone_id: str) -> str:
"""Get detailed information about a milestone including all checkpoints and their status.
Args:
milestone_id: The milestone ID
"""
try:
if not milestone_id:
return "Please provide a milestone_id."
milestone = await client.get_milestone(milestone_id)
return _format_milestone(milestone)
except ClinkApiError as e:
if e.status_code == 404:
return f"Milestone not found: {milestone_id}"
return f"Error: {e}"
@mcp.tool()
async def complete_checkpoint(
milestone_id: str,
order: int,
hil_expiry_seconds: Optional[int] = None,
) -> str:
"""Mark a milestone checkpoint as completed. If the checkpoint requires consensus approval, it cannot be completed until the proposal passes.
Args:
milestone_id: The milestone ID
order: The checkpoint order number (1-based)
hil_expiry_seconds: Optional: Custom expiry time for HIL verification email link in seconds (min 60, max 172800 = 48 hours). Default is 300 (5 minutes).
"""
try:
if not milestone_id:
return "Please provide a milestone_id."
if not order:
return "Please provide the checkpoint order number."
result = await client.complete_checkpoint(
milestone_id,
order,
hil_expiry_seconds=hil_expiry_seconds,
)
# Check if HIL verification is required
if isinstance(result, dict) and is_hil_required(result):
is_pending = result.get("verification_status") == "pending"
was_expired = result.get("previous_verification_expired", False)
if is_pending:
# Existing verification already in progress
output = "Human-in-the-Loop verification already pending.\n\n"
output += "A verification email was already sent and is awaiting human approval.\n"
output += "No new email was sent.\n\n"
output += f"Verification ID: {result.get('verification_id')}\n"
if result.get("expires_at"):
output += f"Expires at: {result['expires_at']}\n"
elif was_expired:
# Previous verification expired, new one sent
output = "Previous verification expired - new email sent.\n\n"
output += "The previous verification link expired without being used.\n"
output += "A new verification email has been sent to the designated approver.\n\n"
output += f"Verification ID: {result.get('verification_id')}\n"
if result.get("expires_in_seconds"):
output += f"Expires in: {result['expires_in_seconds'] // 60} minutes\n"
if result.get("sent_to"):
output += f"Email sent to: {', '.join(result['sent_to'])}\n"
else:
# New verification created
output = "Human-in-the-Loop verification required.\n\n"
output += "A verification email has been sent to the designated approver.\n"
output += "The checkpoint will be marked complete once they verify via the email link.\n\n"
output += f"Verification ID: {result.get('verification_id')}\n"
if result.get("expires_in_seconds"):
output += f"Expires in: {result['expires_in_seconds'] // 60} minutes\n"
if result.get("sent_to"):
output += f"Email sent to: {', '.join(result['sent_to'])}\n"
output += "\nUse list_pending_verifications to see all pending HIL verifications."
return output
milestone = result
completed_checkpoint = None
if milestone.checkpoints:
for cp in milestone.checkpoints:
if cp.order == order:
completed_checkpoint = cp
break
output = f"Checkpoint {order} completed!\n\n"
output += f"Milestone: {milestone.title}\n"
output += f"Progress: {milestone.progress}\n"
output += f"Status: {milestone.status}\n"
if completed_checkpoint:
output += f"\nCompleted checkpoint: {completed_checkpoint.title}"
if completed_checkpoint.git_branch_url:
output += f"\n Branch: {completed_checkpoint.git_branch_url}"
if completed_checkpoint.git_pr_url:
output += f"\n PR: {completed_checkpoint.git_pr_url}"
if completed_checkpoint.git_commit_url:
output += f"\n Commit: {completed_checkpoint.git_commit_url}"
if milestone.status == "closed":
output += "\n\nAll checkpoints completed! Milestone is now closed."
elif milestone.checkpoints:
remaining = [cp for cp in milestone.checkpoints if cp.status == "pending"]
if remaining:
output += "\n\nRemaining checkpoints:\n"
for cp in remaining:
blockers = []
if cp.requires_proposal_id and cp.proposal_status != "approved":
blockers.append("waiting for consensus")
if cp.depends_on:
unmet_deps = []
for dep_order in cp.depends_on:
if isinstance(dep_order, int):
dep_cp = next((c for c in milestone.checkpoints if c.order == dep_order), None)
if dep_cp and dep_cp.status != "completed":
unmet_deps.append(str(dep_order))
if unmet_deps:
blockers.append(f"depends on: {', '.join(unmet_deps)}")
suffix = f" [{'; '.join(blockers)}]" if blockers else ""
output += f" {cp.order}. {cp.title}{suffix}\n"
return output
except ClinkApiError as e:
if e.status_code == 404:
return f"Milestone or checkpoint not found: {milestone_id}/{order}"
if e.status_code == 409:
return "Checkpoint cannot be completed. It may require consensus approval or have incomplete dependencies."
return f"Error: {e}"
@mcp.tool()
async def update_milestone(
milestone_id: str,
title: Optional[str] = None,
description: Optional[str] = None,
) -> str:
"""Update a milestone's title or description. Cannot modify closed milestones.
Args:
milestone_id: The milestone ID
title: New title (optional)
description: New description (optional)
"""
try:
if not milestone_id:
return "Please provide a milestone_id."
if not title and not description:
return "Please provide at least one field to update (title or description)."
milestone = await client.update_milestone(
milestone_id,
title=title,
description=description,
)
output = "Milestone updated!\n\n"
output += _format_milestone(milestone, include_checkpoints=False)
return output
except ClinkApiError as e:
if e.status_code == 404:
return f"Milestone not found: {milestone_id}"
return f"Error: {e}"
@mcp.tool()
async def update_checkpoint(
milestone_id: str,
order: int,
title: Optional[str] = None,
description: Optional[str] = None,
depends_on: Optional[list[int | str]] = None,
git_branch_url: Optional[str] = None,
git_pr_url: Optional[str] = None,
git_commit_url: Optional[str] = None,
) -> str:
"""Update a checkpoint's title, description, dependencies, or git references (branch URL, PR URL, commit URL).
Args:
milestone_id: The milestone ID
order: The checkpoint order number (1-based)
title: New checkpoint title (optional)
description: New checkpoint description (optional)
depends_on: Dependencies. Use number for same-milestone (e.g., 1), or "milestone_id:order" for cross-milestone (e.g., "ms_abc123:4"). Same-milestone deps must reference earlier checkpoints. Cross-milestone deps are validated for cycles.
git_branch_url: Git branch URL (full URL, e.g., https://github.com/org/repo/tree/feature/auth)
git_pr_url: Git pull request URL (full URL, e.g., https://github.com/org/repo/pull/123)
git_commit_url: Git commit URL (full URL, e.g., https://github.com/org/repo/commit/abc123)
"""
try:
if not milestone_id:
return "Please provide a milestone_id."
if not order:
return "Please provide the checkpoint order number."
milestone = await client.update_checkpoint(
milestone_id,
order,
title=title,
description=description,
depends_on=depends_on,
git_branch_url=git_branch_url,
git_pr_url=git_pr_url,
git_commit_url=git_commit_url,
)
output = f"Checkpoint {order} updated!\n\n"
output += _format_milestone(milestone)
return output
except ClinkApiError as e:
if e.status_code == 404:
return f"Milestone or checkpoint not found: {milestone_id}/{order}"
return f"Error: {e}"
@mcp.tool()
async def delete_checkpoint(
milestone_id: str,
order: int,
) -> str:
"""Delete a checkpoint from a milestone. Cannot delete completed checkpoints or checkpoints that others depend on.
Args:
milestone_id: The milestone ID
order: The checkpoint order number (1-based) to delete
"""
try:
if not milestone_id:
return "Please provide a milestone_id."
if not order:
return "Please provide the checkpoint order number."
milestone = await client.delete_checkpoint(milestone_id, order)
output = f"Checkpoint {order} deleted.\n\n"
output += _format_milestone(milestone)
return output
except ClinkApiError as e:
if e.status_code == 404:
return f"Milestone or checkpoint not found: {milestone_id}/{order}"
if e.status_code == 409:
return "Cannot delete this checkpoint. It may be completed or other checkpoints depend on it."
return f"Error: {e}"
@mcp.tool()
async def add_checkpoint(
milestone_id: str,
title: str,
description: Optional[str] = None,
requires_consensus: Optional[bool] = None,
depends_on: Optional[list[int | str]] = None,
git_branch_url: Optional[str] = None,
git_pr_url: Optional[str] = None,
git_commit_url: Optional[str] = None,
position: Optional[int] = None,
) -> str:
"""Add a new checkpoint to an existing milestone. Optionally specify a position to insert at, or it will be appended at the end.
Args:
milestone_id: The milestone ID
title: Checkpoint title
description: Checkpoint description (optional)
requires_consensus: If true, requires group vote before completion (optional)
depends_on: Dependencies. Use number for same-milestone (e.g., 1), or "milestone_id:order" for cross-milestone (e.g., "ms_abc123:4"). Same-milestone deps must reference earlier checkpoints.
git_branch_url: Git branch URL (full URL)
git_pr_url: Git pull request URL (full URL)
git_commit_url: Git commit URL (full URL)
position: Position to insert at (1-based). If not provided, appends at end.
"""
try:
if not milestone_id:
return "Please provide a milestone_id."
if not title:
return "Please provide a checkpoint title."
milestone = await client.add_checkpoint(
milestone_id,
title=title,
description=description,
requires_consensus=requires_consensus,
depends_on=depends_on,
git_branch_url=git_branch_url,
git_pr_url=git_pr_url,
git_commit_url=git_commit_url,
position=position,
)
output = "Checkpoint added!\n\n"
output += _format_milestone(milestone)
return output
except ClinkApiError as e:
if e.status_code == 404:
return f"Milestone not found: {milestone_id}"
return f"Error: {e}"
@mcp.tool()
async def reopen_milestone(milestone_id: str) -> str:
"""Re-open a closed milestone. This allows adding new checkpoints or completing remaining ones.
Args:
milestone_id: The milestone ID to reopen
"""
try:
if not milestone_id:
return "Please provide a milestone_id."
milestone = await client.reopen_milestone(milestone_id)
output = "Milestone reopened!\n\n"
output += _format_milestone(milestone)
return output
except ClinkApiError as e:
if e.status_code == 404:
return f"Milestone not found: {milestone_id}"
if e.status_code == 409:
return "Milestone is already open."
return f"Error: {e}"
