"""
ActiveCollab MCP Server.
"""
from __future__ import annotations
import json
import logging
from datetime import datetime
from mcp.server.fastmcp import FastMCP
from .api.client import ActiveCollabClient
from .utils.config import Config
logging.getLogger("mcp").setLevel(logging.ERROR)
config = Config.load()
client = ActiveCollabClient(config)
mcp = FastMCP("ActiveCollab")
@mcp.tool()
def list_projects() -> str:
"""List all ActiveCollab projects."""
projects = client.list_projects()
result = []
for p in projects:
result.append(f"[{p.get('id')}] {p.get('name')}")
return "\n".join(result)
@mcp.tool()
def get_project(project_id: int) -> str:
"""Get details of a specific project."""
p = client.get_project(project_id)
return json.dumps(p, indent=2)
@mcp.tool()
def list_tasks(project_id: int, status: str = "all") -> str:
"""List tasks for a project. Status can be: all, open, completed."""
tasks = client.list_tasks(project_id)
if status == "open":
tasks = [t for t in tasks if not t.get("is_completed")]
elif status == "completed":
tasks = [t for t in tasks if t.get("is_completed")]
result = []
for t in tasks:
icon = "✓" if t.get("is_completed") else "○"
result.append(f"{icon} [{t.get('id')}] {t.get('name')}")
return "\n".join(result) if result else "No tasks found."
@mcp.tool()
def get_task(project_id: int, task_id: int) -> str:
"""Get details of a specific task."""
t = client.get_task(project_id, task_id)
return json.dumps(t, indent=2)
@mcp.tool()
def my_tasks() -> str:
"""Get all tasks assigned to the current user."""
tasks = client.get_user_tasks()
result = []
for t in tasks:
status = "✓" if t.get("is_completed") else "○"
project = t.get("project_id", "?")
result.append(f"{status} [{t.get('id')}] {t.get('name')} (project: {project})")
return "\n".join(result) if result else "No tasks assigned."
@mcp.tool()
def get_task_comments(task_id: int) -> str:
"""Get all comments on a specific task."""
comments = client.get_task_comments(task_id)
result = []
for c in comments:
author = c.get("created_by_name", "Unknown")
date = c.get("created_on", "")
body = c.get("body", "").strip()
result.append(f"[{date}] {author}:\n{body}\n")
return "\n".join(result) if result else "No comments on this task."
def _format_timestamp(timestamp: int | None, with_time: bool = False) -> str:
if not timestamp:
return ""
fmt = "%Y-%m-%d %H:%M" if with_time else "%Y-%m-%d"
return datetime.fromtimestamp(timestamp).strftime(fmt)
_user_cache: dict[int, str] = {}
def _get_user_name(user_id: int) -> str:
if not _user_cache:
users = client._get("/users")
for u in users:
_user_cache[u["id"]] = u.get("display_name", "Unknown")
return _user_cache.get(user_id, "Unknown")
@mcp.tool()
def get_task_time_records(project_id: int, task_id: int) -> str:
"""Get all time records for a specific task."""
records = client.get_task_time_records(project_id, task_id)
result = []
total_hours = 0.0
for r in records:
user = r.get("user_name", "Unknown")
hours = r.get("value", 0)
total_hours += float(hours)
date = _format_timestamp(r.get("record_date"))
summary = r.get("summary", "").strip()
result.append(f"[{date}] {user}: {hours}h - {summary}")
if result:
result.append(f"\nTotal: {total_hours}h")
return "\n".join(result)
return "No time records for this task."
@mcp.tool()
def get_project_time_records(project_id: int) -> str:
"""Get all time records for a project."""
records = client.get_project_time_records(project_id)
result = []
total_hours = 0.0
for r in records:
user = r.get("user_name", "Unknown")
hours = r.get("value", 0)
total_hours += float(hours)
date = _format_timestamp(r.get("record_date"))
summary = r.get("summary", "").strip()
task_id = r.get("parent_id", "")
result.append(f"[{date}] {user}: {hours}h (task #{task_id}) - {summary}")
if result:
result.append(f"\nTotal: {total_hours}h")
return "\n".join(result)
return "No time records for this project."
@mcp.tool()
def get_notifications() -> str:
"""Get recent notifications for the current user."""
notifications = client.get_notifications()
result = []
for n in notifications[:20]:
sender = _get_user_name(n.get("sender_id", 0))
parent_type = n.get("parent_type", "")
parent_id = n.get("parent_id", "")
notif_type = n.get("class", "").replace("Notification", "")
date = _format_timestamp(n.get("created_on"), with_time=True)
result.append(f"[{date}] {notif_type} from {sender} on {parent_type} #{parent_id}")
return "\n".join(result) if result else "No notifications."
@mcp.tool()
def get_subtasks(project_id: int, task_id: int) -> str:
"""Get all subtasks for a specific task."""
subtasks = client.get_subtasks(project_id, task_id)
result = []
open_count = 0
completed_count = 0
for s in subtasks:
icon = "✓" if s.get("is_completed") else "○"
if s.get("is_completed"):
completed_count += 1
else:
open_count += 1
name = s.get("name", "")
assignee_id = s.get("assignee_id", 0)
assignee = _get_user_name(assignee_id) if assignee_id else "Unassigned"
due = _format_timestamp(s.get("due_on"))
due_str = f" (due: {due})" if due else ""
result.append(f"{icon} [{s.get('id')}] {name} - {assignee}{due_str}")
if result:
result.insert(0, f"Subtasks: {completed_count} completed, {open_count} open\n")
return "\n".join(result)
return "No subtasks for this task."
@mcp.tool()
def get_overdue_tasks(project_id: int) -> str:
"""Get all tasks with overdue deadlines for a project."""
tasks = client.list_tasks(project_id)
now = datetime.now().timestamp()
overdue = []
for t in tasks:
if t.get("is_completed"):
continue
due_on = t.get("due_on")
if due_on and due_on < now:
days_overdue = int((now - due_on) / 86400)
due_date = _format_timestamp(due_on)
overdue.append({
"task": t,
"due_on": due_on,
"days_overdue": days_overdue,
"due_date": due_date
})
overdue.sort(key=lambda x: x["due_on"])
result = []
for item in overdue:
t = item["task"]
result.append(f"[{item['due_date']}] ({item['days_overdue']}d overdue) #{t.get('id')} {t.get('name')}")
if result:
result.insert(0, f"Found {len(overdue)} overdue tasks:\n")
return "\n".join(result)
return "No overdue tasks."
@mcp.tool()
def get_tasks_with_deadlines(project_id: int) -> str:
"""Get all open tasks with deadlines, sorted by due date."""
tasks = client.list_tasks(project_id)
now = datetime.now().timestamp()
with_deadline = []
for t in tasks:
if t.get("is_completed"):
continue
due_on = t.get("due_on")
if due_on:
days_until = int((due_on - now) / 86400)
due_date = _format_timestamp(due_on)
with_deadline.append({
"task": t,
"due_on": due_on,
"days_until": days_until,
"due_date": due_date
})
with_deadline.sort(key=lambda x: x["due_on"])
result = []
for item in with_deadline:
t = item["task"]
days = item["days_until"]
if days < 0:
status = f"{abs(days)}d OVERDUE"
elif days == 0:
status = "DUE TODAY"
else:
status = f"in {days}d"
result.append(f"[{item['due_date']}] ({status}) #{t.get('id')} {t.get('name')}")
if result:
result.insert(0, f"Found {len(with_deadline)} tasks with deadlines:\n")
return "\n".join(result)
return "No tasks with deadlines."
@mcp.tool()
def get_labels() -> str:
"""Get all available task labels (global labels only)."""
labels = client.get_labels()
task_labels = [l for l in labels if l.get("class") == "TaskLabel" and l.get("is_global")]
task_labels.sort(key=lambda x: x.get("name", ""))
result = []
for l in task_labels:
result.append(f"[{l.get('id')}] {l.get('name')}")
if result:
result.insert(0, f"Found {len(task_labels)} global task labels:\n")
return "\n".join(result)
return "No labels found."
@mcp.tool()
def get_tasks_by_label(project_id: int, label_id: int) -> str:
"""Get all tasks with a specific label."""
tasks = client.list_tasks(project_id)
matching = []
label_name = None
for t in tasks:
task_labels = t.get("labels", [])
for lbl in task_labels:
if lbl.get("id") == label_id:
matching.append(t)
if not label_name:
label_name = lbl.get("name")
break
result = []
for t in matching:
icon = "✓" if t.get("is_completed") else "○"
result.append(f"{icon} [{t.get('id')}] {t.get('name')}")
if result:
header = f"Found {len(matching)} tasks with label"
if label_name:
header += f" '{label_name}'"
result.insert(0, f"{header}:\n")
return "\n".join(result)
return f"No tasks found with label #{label_id}."
@mcp.tool()
def get_task_lists(project_id: int) -> str:
"""Get all task lists for a project."""
task_lists = client.get_task_lists(project_id)
task_lists.sort(key=lambda x: x.get("position", 0))
result = []
for tl in task_lists:
open_count = tl.get("open_tasks", 0)
completed_count = tl.get("completed_tasks", 0)
result.append(f"[{tl.get('id')}] {tl.get('name')} ({open_count} open, {completed_count} completed)")
if result:
result.insert(0, f"Found {len(task_lists)} task lists:\n")
return "\n".join(result)
return "No task lists found."
@mcp.tool()
def get_tasks_by_list(project_id: int, task_list_id: int) -> str:
"""Get all tasks in a specific task list."""
tasks = client.list_tasks(project_id)
task_lists = client.get_task_lists(project_id)
list_name = None
for tl in task_lists:
if tl.get("id") == task_list_id:
list_name = tl.get("name")
break
matching = [t for t in tasks if t.get("task_list_id") == task_list_id]
result = []
for t in matching:
icon = "✓" if t.get("is_completed") else "○"
result.append(f"{icon} [{t.get('id')}] {t.get('name')}")
if result:
header = f"Found {len(matching)} tasks"
if list_name:
header += f" in '{list_name}'"
result.insert(0, f"{header}:\n")
return "\n".join(result)
return f"No tasks found in task list #{task_list_id}."
@mcp.tool()
def get_users() -> str:
"""Get all users (members and owners only, not clients)."""
users = client.get_users()
team_members = [u for u in users if u.get("class") in ("Member", "Owner")]
team_members.sort(key=lambda x: x.get("display_name", ""))
result = []
for u in team_members:
role = "👑" if u.get("class") == "Owner" else "👤"
result.append(f"{role} [{u.get('id')}] {u.get('display_name')} ({u.get('email')})")
if result:
result.insert(0, f"Found {len(team_members)} team members:\n")
return "\n".join(result)
return "No users found."
@mcp.tool()
def get_tasks_by_assignee(project_id: int, user_id: int) -> str:
"""Get all tasks assigned to a specific user in a project."""
tasks = client.list_tasks(project_id)
user_name = _get_user_name(user_id)
matching = [t for t in tasks if t.get("assignee_id") == user_id]
open_tasks = [t for t in matching if not t.get("is_completed")]
completed_tasks = [t for t in matching if t.get("is_completed")]
result = []
if open_tasks:
result.append(f"Open ({len(open_tasks)}):")
for t in open_tasks:
result.append(f" ○ [{t.get('id')}] {t.get('name')}")
if completed_tasks:
result.append(f"\nCompleted ({len(completed_tasks)}):")
for t in completed_tasks[:10]:
result.append(f" ✓ [{t.get('id')}] {t.get('name')}")
if len(completed_tasks) > 10:
result.append(f" ... and {len(completed_tasks) - 10} more")
if result:
header = f"Tasks assigned to {user_name}:\n"
result.insert(0, header)
return "\n".join(result)
return f"No tasks assigned to user #{user_id}."
if __name__ == "__main__":
mcp.run(transport="stdio")
