"""Consensus/voting proposal tools."""
from typing import Optional
from mcp.server.fastmcp import FastMCP
from .. import client
from ..types import ClinkApiError, Proposal, is_hil_required
def _format_proposal(p: Proposal, include_votes: bool = False) -> str:
"""Format a proposal for display."""
output = f"Title: {p.title}\n"
output += f"ID: {p.proposal_id}\n"
output += f"Status: {p.status}\n"
output += f"Type: {p.voting_type}\n"
output += f"Threshold: {p.threshold_type}\n"
if p.description:
output += f"Description: {p.description}\n"
if p.options:
output += f"Options: {', '.join(p.options)}\n"
if p.deadline:
output += f"Deadline: {p.deadline}\n"
if p.vote_count is not None:
output += f"Votes: {p.vote_count}\n"
if p.tally:
tally_str = ", ".join(f"{k}: {v}" for k, v in p.tally.items())
output += f"Tally: {tally_str}\n"
if p.status == "finalized":
output += f"Result: {p.result or 'No result'}\n"
output += f"Threshold met: {'Yes' if p.threshold_met else 'No'}\n"
if include_votes and p.votes:
output += "\nVotes:\n"
for v in p.votes:
comment = f" - {v.comment}" if v.comment else ""
output += f" * {v.voter_id}: {v.vote}{comment}\n"
return output
def register(mcp: FastMCP) -> None:
"""Register consensus/voting tools with the MCP server."""
@mcp.tool()
async def create_proposal(
group: str,
title: str,
description: Optional[str] = None,
voting_type: Optional[str] = None,
threshold_type: Optional[str] = None,
options: Optional[list[str]] = None,
deadline_hours: Optional[int] = None,
) -> str:
"""Create a voting proposal for group decision-making. Supports different voting types (yes/no, single choice, ranked) and threshold requirements.
Args:
group: The group slug (e.g., "backend-team") or group ID
title: Proposal title
description: Proposal description (optional)
voting_type: Type of voting: yes_no (approve/reject), single (choose one option), ranked (preference order). Default: yes_no
threshold_type: Required threshold: majority (>50%), two_thirds (>=66%), unanimous (100%), quorum (>50% participation + majority). Default: majority
options: Options for single/ranked voting (required for those types, ignored for yes_no)
deadline_hours: Optional deadline in hours from now
"""
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."
if not title:
return "Please provide a proposal title."
# Validate voting type and options
if voting_type in ("single", "ranked") and not options:
return f"Options are required for {voting_type} voting type."
group_id = await client.resolve_group(group)
proposal = await client.create_proposal(
group_id,
title=title,
description=description,
voting_type=voting_type,
threshold_type=threshold_type,
options=options,
deadline_hours=deadline_hours,
)
output = "Proposal created successfully!\n\n"
output += _format_proposal(proposal)
return output
except ClinkApiError as e:
return f"Error: {e}"
@mcp.tool()
async def list_proposals(
group: str,
status: Optional[str] = None,
limit: Optional[int] = None,
) -> str:
"""List voting proposals for a Clink group. Shows status and vote counts.
Args:
group: The group slug (e.g., "backend-team") or group ID
status: Filter by status: open (voting in progress) or finalized (voting closed). Default: all.
limit: Maximum proposals 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)
proposals = await client.list_proposals(
group_id,
status=status,
limit=limit,
)
if not proposals:
return f"No proposals found in group '{group}'."
output = f"Proposals in {group} ({len(proposals)}):\n\n"
for p in proposals:
votes = f" | Votes: {p.vote_count}" if p.vote_count is not None else ""
output += f"* {p.title}\n"
output += f" ID: {p.proposal_id}\n"
output += f" Status: {p.status} | Type: {p.voting_type}{votes}\n\n"
return output.strip()
except ClinkApiError as e:
return f"Error: {e}"
@mcp.tool()
async def get_proposal(proposal_id: str) -> str:
"""Get detailed information about a proposal including all votes and their comments.
Args:
proposal_id: The proposal ID
"""
try:
if not proposal_id:
return "Please provide a proposal_id."
proposal = await client.get_proposal(proposal_id)
return _format_proposal(proposal, include_votes=True)
except ClinkApiError as e:
if e.status_code == 404:
return f"Proposal not found: {proposal_id}"
return f"Error: {e}"
@mcp.tool()
async def cast_vote(
proposal_id: str,
vote: str,
comment: Optional[str] = None,
hil_expiry_seconds: Optional[int] = None,
) -> str:
"""Cast a vote on a proposal. For yes/no proposals, vote "yes" or "no". For single choice, vote with the option name. Include a comment to explain your reasoning.
Args:
proposal_id: The proposal ID
vote: Your vote: "yes" or "no" for yes_no proposals, option name for single choice, comma-separated preferences for ranked
comment: Optional comment explaining your vote
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 proposal_id:
return "Please provide a proposal_id."
if not vote:
return "Please provide your vote."
result = await client.cast_vote(
proposal_id,
vote=vote,
comment=comment,
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"Vote: {vote}\n"
if comment:
output += f"Comment: {comment}\n"
output += f"\nVerification 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"Vote: {vote}\n"
if comment:
output += f"Comment: {comment}\n"
output += f"\nVerification 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 += "Your vote will be recorded once they verify via the email link.\n\n"
output += f"Vote: {vote}\n"
if comment:
output += f"Comment: {comment}\n"
output += f"\nVerification 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
proposal = result
output = "Vote recorded!\n\n"
output += f"Proposal: {proposal.title}\n"
output += f"Your vote: {vote}\n"
if comment:
output += f"Comment: {comment}\n"
# Show current tally
if proposal.tally:
output += "\nCurrent tally:\n"
for option, count in proposal.tally.items():
output += f" {option}: {count}\n"
output += f"\nTotal votes: {proposal.vote_count}"
return output
except ClinkApiError as e:
if e.status_code == 404:
return f"Proposal not found: {proposal_id}"
if e.status_code == 409:
return "Cannot vote on this proposal. It may be finalized or you may have already voted."
if e.status_code == 400:
return f"Invalid vote. Check the voting type and available options. Error: {e}"
return f"Error: {e}"
@mcp.tool()
async def finalize_proposal(
proposal_id: str,
total_eligible_voters: Optional[int] = None,
) -> str:
"""Close voting on a proposal and compute the final result. The result depends on the threshold type (majority, two-thirds, unanimous, quorum).
Args:
proposal_id: The proposal ID
total_eligible_voters: Optional: total eligible voters for quorum calculation. If not provided, quorum is calculated based on votes cast.
"""
try:
if not proposal_id:
return "Please provide a proposal_id."
proposal = await client.finalize_proposal(
proposal_id,
total_eligible_voters=total_eligible_voters,
)
output = "Proposal finalized!\n\n"
output += _format_proposal(proposal, include_votes=True)
return output
except ClinkApiError as e:
if e.status_code == 404:
return f"Proposal not found: {proposal_id}"
if e.status_code == 409:
return "Proposal is already finalized."
return f"Error: {e}"
