"""Comment and activity tools for OpenProject MCP server."""
from typing import Any
from ..client import OpenProjectClient
from ..utils.hal import build_formattable, get_embedded_collection
def _format_activity_markdown(activity: dict[str, Any]) -> str:
"""Format a single activity as markdown.
Args:
activity: Activity object from OpenProject API
Returns:
Formatted markdown string
"""
activity_id = activity.get("id", "N/A")
activity_type = activity.get("_type", "Activity")
version = activity.get("version", "N/A")
internal = activity.get("internal", False)
created_at = activity.get("createdAt", "N/A")
updated_at = activity.get("updatedAt", "N/A")
# Get comment text
comment_obj = activity.get("comment", {})
comment_text = comment_obj.get("raw", "") if comment_obj else ""
# Get user information
user_link = activity.get("_links", {}).get("user", {})
user_title = user_link.get("title", "Unknown User")
# Get details (changes)
details = activity.get("details", [])
details_summary = []
for detail in details[:3]: # Show first 3 details
if isinstance(detail, dict):
detail_format = detail.get("format", "")
detail_html = detail.get("html", "")
if detail_html:
details_summary.append(f" - {detail_html}")
markdown = f"""
### Activity #{activity_id} - {activity_type}
**Created:** {created_at}
"""
if comment_text:
markdown += f"\n**Comment:**\n```\n{comment_text}\n```\n"
if details_summary:
markdown += f"\n**Changes:**\n" + "\n".join(details_summary) + "\n"
return markdown
async def get_work_package_activities(
work_package_id: int, page: int = 1, page_size: int = 20
) -> str:
"""Retrieve all activities and comments for a work package.
Args:
work_package_id: Work package ID
page: Page number (default: 1)
page_size: Items per page (default: 20)
Returns:
Formatted markdown string with activities summary
"""
client = OpenProjectClient()
try:
params = {
"pageSize": page_size,
"offset": (page - 1) * page_size,
}
result = await client.get(
f"work_packages/{work_package_id}/activities", params=params
)
# Extract metadata
total = result.get("total", 0)
count = result.get("count", 0)
# Get activities
activities = get_embedded_collection(result, "elements")
# Format as markdown
markdown = f"""# Work Package #{work_package_id} Activities
**Total Activities:** {total}
**Showing:** {count} (Page {page})
---
"""
if not activities:
markdown += "\n*No activities found.*\n"
else:
for activity in activities:
markdown += _format_activity_markdown(activity)
markdown += "\n---\n"
return markdown
finally:
await client.close()
async def create_comment(
work_package_id: int, comment: str, internal: bool = False
) -> dict[str, Any]:
"""Add a comment to a work package.
Args:
work_package_id: Work package ID
comment: Comment text in markdown format
internal: Whether the comment is internal (default: False)
Returns:
Created activity object containing the comment
"""
client = OpenProjectClient()
try:
payload = {
"comment": build_formattable(comment),
}
# Internal comments might be handled differently depending on OpenProject version
# Adding it to the payload if requested
if internal:
payload["internal"] = internal
result = await client.post(
f"work_packages/{work_package_id}/activities", data=payload
)
return result
finally:
await client.close()
async def get_activity(activity_id: int) -> dict[str, Any]:
"""Get details of a specific activity.
Args:
activity_id: Activity ID
Returns:
Activity object with details
"""
client = OpenProjectClient()
try:
result = await client.get(f"activities/{activity_id}")
return result
finally:
await client.close()
async def update_comment(
activity_id: int, comment: str, lock_version: int
) -> dict[str, Any]:
"""Update an existing comment.
Args:
activity_id: Activity ID of the comment to update
comment: New comment text in markdown format
lock_version: Current lock version (get from activity first)
Returns:
Updated activity object
"""
client = OpenProjectClient()
try:
payload = {
"comment": build_formattable(comment),
"lockVersion": lock_version,
}
result = await client.patch(f"activities/{activity_id}", data=payload)
return result
finally:
await client.close()
