"""
Clink API Client
HTTP client for communicating with the Clink API using API key authentication.
"""
import os
import re
from typing import Any, Optional, Union
from urllib.parse import urlencode, quote
from uuid import uuid4
import httpx
from .types import (
Clink,
ClaimResponse,
ClinkApiError,
CompleteResponse,
FeedbackCategory,
FeedbackResponse,
Group,
Member,
Milestone,
MilestoneStatus,
PendingVerification,
PermissionsResponse,
Project,
ProjectStatus,
Proposal,
ReleaseResponse,
is_hil_required,
)
# Configuration from environment
API_URL = os.environ.get("CLINK_API_URL", "https://api.clink.voxos.ai")
API_KEY = os.environ.get("CLINK_API_KEY")
# Session ID - unique per MCP server instance
# Used to distinguish concurrent agents using the same API key
SESSION_ID = os.environ.get("CLINK_SESSION_ID", str(uuid4()))
# UUID regex pattern
UUID_PATTERN = re.compile(
r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
re.IGNORECASE,
)
def is_configured() -> bool:
"""Check if API key is configured."""
return bool(API_KEY and API_KEY.startswith("sk_live_"))
def get_config_status() -> str:
"""Get configuration status message."""
if not API_KEY:
return "CLINK_API_KEY environment variable is not set. Get your API key from https://app.clink.voxos.ai/settings"
if not API_KEY.startswith("sk_live_"):
return 'CLINK_API_KEY must start with "sk_live_". Check your API key.'
return "Configured"
async def request(
method: str,
path: str,
body: Optional[dict[str, Any]] = None,
) -> Any:
"""Make an authenticated request to the Clink API."""
if not is_configured():
raise ClinkApiError(get_config_status(), 401)
url = f"{API_URL}{path}"
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
"X-Clink-Session-Id": SESSION_ID,
}
try:
async with httpx.AsyncClient() as client:
response = await client.request(
method=method,
url=url,
headers=headers,
json=body,
timeout=30.0,
)
data = response.json()
if not response.is_success or not data.get("success"):
raise ClinkApiError(
data.get("message", f"Request failed: {response.reason_phrase}"),
response.status_code,
data.get("error"),
)
return data.get("data")
except httpx.HTTPError as e:
raise ClinkApiError(f"Failed to connect to Clink API: {e}", 0)
def is_uuid(s: str) -> bool:
"""Check if a string looks like a UUID."""
return bool(UUID_PATTERN.match(s))
# ============================================================================
# Groups API
# ============================================================================
async def list_groups() -> list[Group]:
"""List all groups the user belongs to."""
response = await request("GET", "/groups")
return [Group(**g) for g in response.get("groups", [])]
async def resolve_group(group_ref: str) -> str:
"""
Resolve a group reference (slug or ID) to a group ID.
If it looks like a UUID, returns it directly.
Otherwise, looks up the group by slug.
"""
if is_uuid(group_ref):
return group_ref
# Look up by slug
response = await request("GET", f"/groups?slug={quote(group_ref)}")
groups = response.get("groups", [])
if not groups:
raise ClinkApiError(
f'Group not found: "{group_ref}". Use list_groups to see available groups.',
404,
)
return groups[0]["group_id"]
async def get_group(group_id: str) -> Group:
"""Get details of a specific group."""
response = await request("GET", f"/groups/{group_id}")
return Group(**response)
async def list_members(group_id: str) -> list[Member]:
"""List members of a group."""
response = await request("GET", f"/groups/{group_id}/members")
return [Member(**m) for m in response.get("members", [])]
# ============================================================================
# Clinks API
# ============================================================================
async def get_clinks(
*,
group_id: Optional[str] = None,
since: Optional[str] = None,
limit: Optional[int] = None,
status: Optional[str] = None,
for_me: Optional[bool] = None,
unread_only: Optional[bool] = None,
mark_read: Optional[bool] = None,
) -> list[Clink]:
"""Get clinks with optional filters."""
params: dict[str, Any] = {}
if group_id:
params["group_id"] = group_id
if since:
params["since"] = since
if limit:
params["limit"] = limit
if status:
params["status"] = status
if for_me:
params["for_me"] = "true"
if unread_only:
params["unread_only"] = "true"
if mark_read:
params["mark_read"] = "true"
query = urlencode(params) if params else ""
path = f"/clinks?{query}" if query else "/clinks"
response = await request("GET", path)
return [Clink(**c) for c in response.get("clinks", [])]
async def mark_read(group_ids: list[str]) -> dict[str, Any]:
"""Mark clinks in groups as read."""
return await request("POST", "/clinks/mark-read", {"group_ids": group_ids})
async def send_clink(
group_id: str,
content: str,
*,
for_recipient: Optional[str] = None,
) -> Clink:
"""Send a clink to a group."""
body: dict[str, str] = {"content": content}
if for_recipient:
body["for"] = for_recipient
response = await request("POST", f"/groups/{group_id}/clinks", body)
return Clink(**response)
# ============================================================================
# Inbox API (with claim/ack support)
# ============================================================================
async def check_inbox(
*,
status: Optional[str] = None,
for_me: Optional[bool] = None,
claim: Optional[bool] = None,
limit: Optional[int] = None,
since: Optional[str] = None,
) -> dict[str, Any]:
"""Check inbox with claim/ack support."""
params: dict[str, Any] = {}
if status:
params["status"] = status
if for_me is not None:
params["for_me"] = str(for_me).lower()
if claim:
params["claim"] = "true"
if limit:
params["limit"] = limit
if since:
params["since"] = since
query = urlencode(params) if params else ""
path = f"/inbox?{query}" if query else "/inbox"
response = await request("GET", path)
# Return the raw response with clinks parsed
return {
"clinks": [Clink(**c) for c in response.get("clinks", [])],
"count": response.get("count", 0),
"claimed_count": response.get("claimed_count"),
}
# ============================================================================
# Claim/Ack API
# ============================================================================
async def claim_clink(
clink_id: str,
*,
timeout_seconds: Optional[int] = None,
) -> Clink:
"""Claim a clink for processing."""
body: dict[str, int] = {}
if timeout_seconds:
body["timeout_seconds"] = timeout_seconds
response = await request("POST", f"/clinks/{clink_id}/claim", body)
return Clink(**response)
async def release_clink(clink_id: str) -> Clink:
"""Release a claimed clink without completing."""
response = await request("POST", f"/clinks/{clink_id}/release", {})
return Clink(**response)
async def complete_clink(
clink_id: str,
*,
response_text: Optional[str] = None,
) -> CompleteResponse:
"""Complete a claimed clink."""
body: dict[str, str] = {}
if response_text:
body["response"] = response_text
response = await request("POST", f"/clinks/{clink_id}/complete", body)
return CompleteResponse(
clink=Clink(**response["clink"]),
response_clink_id=response.get("response_clink_id"),
)
# ============================================================================
# Milestones API
# ============================================================================
async def create_milestone(
group_id: str,
*,
title: str,
checkpoints: list[dict[str, Any]],
description: Optional[str] = None,
project_id: Optional[str] = None,
) -> Milestone:
"""Create a milestone with checkpoints."""
body: dict[str, Any] = {
"title": title,
"checkpoints": checkpoints,
}
if description:
body["description"] = description
if project_id:
body["project_id"] = project_id
response = await request("POST", f"/groups/{group_id}/milestones", body)
return Milestone(**response)
async def list_milestones(
group_id: str,
*,
status: Optional[MilestoneStatus] = None,
limit: Optional[int] = None,
) -> list[Milestone]:
"""List milestones for a group."""
params: dict[str, Any] = {}
if status:
params["status"] = status
if limit:
params["limit"] = limit
query = urlencode(params) if params else ""
path = f"/groups/{group_id}/milestones?{query}" if query else f"/groups/{group_id}/milestones"
response = await request("GET", path)
return [Milestone(**m) for m in response.get("milestones", [])]
async def get_milestone(milestone_id: str) -> Milestone:
"""Get a milestone with its checkpoints."""
response = await request("GET", f"/milestones/{milestone_id}")
return Milestone(**response)
async def update_milestone(
milestone_id: str,
*,
title: Optional[str] = None,
description: Optional[str] = None,
) -> Milestone:
"""Update a milestone."""
body: dict[str, str] = {}
if title:
body["title"] = title
if description:
body["description"] = description
response = await request("PATCH", f"/milestones/{milestone_id}", body)
return Milestone(**response)
async def complete_checkpoint(
milestone_id: str,
order: int,
*,
hil_expiry_seconds: Optional[int] = None,
) -> Union[Milestone, dict[str, Any]]:
"""Complete a checkpoint. Returns Milestone on success, or dict with HIL info if verification required."""
body: dict[str, Any] = {}
if hil_expiry_seconds is not None:
body["hil_expiry_seconds"] = hil_expiry_seconds
response = await request(
"POST",
f"/milestones/{milestone_id}/checkpoints/{order}/complete",
body,
)
# Check if HIL verification is required - return raw dict
if is_hil_required(response):
return response
return Milestone(**response)
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,
) -> Milestone:
"""Update a checkpoint."""
body: dict[str, Any] = {}
if title:
body["title"] = title
if description:
body["description"] = description
if depends_on is not None:
body["depends_on"] = depends_on
if git_branch_url:
body["git_branch_url"] = git_branch_url
if git_pr_url:
body["git_pr_url"] = git_pr_url
if git_commit_url:
body["git_commit_url"] = git_commit_url
response = await request(
"PATCH",
f"/milestones/{milestone_id}/checkpoints/{order}",
body,
)
return Milestone(**response)
async def delete_checkpoint(milestone_id: str, order: int) -> Milestone:
"""Delete a checkpoint."""
response = await request(
"DELETE",
f"/milestones/{milestone_id}/checkpoints/{order}",
)
return Milestone(**response)
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,
) -> Milestone:
"""Add a checkpoint to an existing milestone."""
body: dict[str, Any] = {"title": title}
if description:
body["description"] = description
if requires_consensus is not None:
body["requires_consensus"] = requires_consensus
if depends_on is not None:
body["depends_on"] = depends_on
if git_branch_url:
body["git_branch_url"] = git_branch_url
if git_pr_url:
body["git_pr_url"] = git_pr_url
if git_commit_url:
body["git_commit_url"] = git_commit_url
if position is not None:
body["position"] = position
response = await request("POST", f"/milestones/{milestone_id}/checkpoints", body)
return Milestone(**response)
async def reopen_milestone(milestone_id: str) -> Milestone:
"""Re-open a closed milestone."""
response = await request("POST", f"/milestones/{milestone_id}/reopen", {})
return Milestone(**response)
# ============================================================================
# Projects API
# ============================================================================
async def create_project(
group_id: str,
*,
title: str,
description: Optional[str] = None,
slug: Optional[str] = None,
color: Optional[str] = None,
) -> Project:
"""Create a project in a group."""
body: dict[str, str] = {"title": title}
if description:
body["description"] = description
if slug:
body["slug"] = slug
if color:
body["color"] = color
response = await request("POST", f"/groups/{group_id}/projects", body)
return Project(**response)
async def list_projects(
group_id: str,
*,
status: Optional[ProjectStatus] = None,
limit: Optional[int] = None,
) -> list[Project]:
"""List projects for a group."""
params: dict[str, Any] = {}
if status:
params["status"] = status
if limit:
params["limit"] = limit
query = urlencode(params) if params else ""
path = f"/groups/{group_id}/projects?{query}" if query else f"/groups/{group_id}/projects"
response = await request("GET", path)
return [Project(**p) for p in response.get("projects", [])]
async def get_project(project_id: str) -> Project:
"""Get a project by ID."""
response = await request("GET", f"/projects/{project_id}")
return Project(**response)
async def update_project(
project_id: str,
*,
title: Optional[str] = None,
description: Optional[str] = None,
slug: Optional[str] = None,
color: Optional[str] = None,
) -> Project:
"""Update a project."""
body: dict[str, str] = {}
if title:
body["title"] = title
if description:
body["description"] = description
if slug:
body["slug"] = slug
if color:
body["color"] = color
response = await request("PATCH", f"/projects/{project_id}", body)
return Project(**response)
async def complete_project(project_id: str) -> Project:
"""Mark a project as completed."""
response = await request("POST", f"/projects/{project_id}/complete", {})
return Project(**response)
async def archive_project(project_id: str) -> Project:
"""Archive a project."""
response = await request("POST", f"/projects/{project_id}/archive", {})
return Project(**response)
async def reopen_project(project_id: str) -> Project:
"""Reopen a completed or archived project."""
response = await request("POST", f"/projects/{project_id}/reopen", {})
return Project(**response)
async def list_project_milestones(
project_id: str,
*,
status: Optional[MilestoneStatus] = None,
limit: Optional[int] = None,
) -> list[Milestone]:
"""List milestones in a project."""
params: dict[str, Any] = {}
if status:
params["status"] = status
if limit:
params["limit"] = limit
query = urlencode(params) if params else ""
path = f"/projects/{project_id}/milestones?{query}" if query else f"/projects/{project_id}/milestones"
response = await request("GET", path)
return [Milestone(**m) for m in response.get("milestones", [])]
# ============================================================================
# Consensus/Proposals API
# ============================================================================
async def create_proposal(
group_id: 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,
) -> Proposal:
"""Create a voting proposal."""
body: dict[str, Any] = {"title": title}
if description:
body["description"] = description
if voting_type:
body["voting_type"] = voting_type
if threshold_type:
body["threshold_type"] = threshold_type
if options:
body["options"] = options
if deadline_hours:
body["deadline_hours"] = deadline_hours
response = await request("POST", f"/groups/{group_id}/proposals", body)
return Proposal(**response)
async def list_proposals(
group_id: str,
*,
status: Optional[str] = None,
limit: Optional[int] = None,
) -> list[Proposal]:
"""List proposals for a group."""
params: dict[str, Any] = {}
if status:
params["status"] = status
if limit:
params["limit"] = limit
query = urlencode(params) if params else ""
path = f"/groups/{group_id}/proposals?{query}" if query else f"/groups/{group_id}/proposals"
response = await request("GET", path)
return [Proposal(**p) for p in response.get("proposals", [])]
async def get_proposal(proposal_id: str) -> Proposal:
"""Get a proposal with votes."""
response = await request("GET", f"/proposals/{proposal_id}")
return Proposal(**response)
async def cast_vote(
proposal_id: str,
*,
vote: str,
comment: Optional[str] = None,
hil_expiry_seconds: Optional[int] = None,
) -> Union[Proposal, dict[str, Any]]:
"""Cast a vote on a proposal. Returns Proposal on success, or dict with HIL info if verification required."""
body: dict[str, Any] = {"vote": vote}
if comment:
body["comment"] = comment
if hil_expiry_seconds is not None:
body["hil_expiry_seconds"] = hil_expiry_seconds
response = await request("POST", f"/proposals/{proposal_id}/vote", body)
# Check if HIL verification is required - return raw dict
if is_hil_required(response):
return response
return Proposal(**response)
async def finalize_proposal(
proposal_id: str,
*,
total_eligible_voters: Optional[int] = None,
) -> Proposal:
"""Finalize a proposal and compute result."""
body: dict[str, int] = {}
if total_eligible_voters is not None:
body["total_eligible_voters"] = total_eligible_voters
response = await request("POST", f"/proposals/{proposal_id}/finalize", body)
return Proposal(**response)
# ============================================================================
# Feedback API
# ============================================================================
async def submit_feedback(
*,
category: FeedbackCategory,
content: str,
tool: Optional[str] = None,
) -> FeedbackResponse:
"""
Submit feedback to help improve Clink.
Requires an API key with feedback permission enabled.
"""
body: dict[str, Any] = {
"category": category,
"content": content,
}
if tool:
body["context"] = {"tool": tool}
response = await request("POST", "/feedback", body)
return FeedbackResponse(**response)
# ============================================================================
# Permissions API
# ============================================================================
async def get_my_permissions() -> PermissionsResponse:
"""Get the permissions for the currently authenticated API key."""
response = await request("GET", "/api-keys/me/permissions")
return PermissionsResponse(**response)
# ============================================================================
# HIL / Verification API
# ============================================================================
async def list_pending_verifications(
group_id: str,
*,
limit: Optional[int] = None,
) -> list[PendingVerification]:
"""
List pending HIL verifications for a group.
Useful for agents to see what's awaiting human approval.
"""
params: dict[str, Any] = {}
if limit:
params["limit"] = limit
query = urlencode(params) if params else ""
path = f"/groups/{group_id}/hil/pending?{query}" if query else f"/groups/{group_id}/hil/pending"
response = await request("GET", path)
return [PendingVerification(**v) for v in response.get("verifications", [])]
