"""Work package management tools for OpenProject MCP server."""
from typing import Any
from ..client import OpenProjectClient
from ..utils.hal import build_formattable, build_link, get_embedded_collection
async def create_work_package(
project_id: str,
subject: str,
description: str | None = None,
type_id: int = 1,
status_id: int | None = None,
priority_id: int | None = None,
assignee_id: int | None = None,
notify: bool = True,
) -> dict[str, Any]:
"""Create a new work package in a project.
Args:
project_id: Project identifier or ID
subject: Work package title (required)
description: Work package description in markdown format
type_id: Type ID (default: 1 for Task)
status_id: Status ID (optional, uses project default if not provided)
priority_id: Priority ID (optional, uses default if not provided)
assignee_id: User ID to assign the work package to
notify: Whether to send notifications (default: True)
Returns:
Created work package object with ID, subject, and all properties
"""
client = OpenProjectClient()
try:
payload: dict[str, Any] = {
"subject": subject,
"_links": {
"type": build_link(f"/api/v3/types/{type_id}"),
},
}
if description:
payload["description"] = build_formattable(description)
if status_id:
payload["_links"]["status"] = build_link(f"/api/v3/statuses/{status_id}")
if priority_id:
payload["_links"]["priority"] = build_link(
f"/api/v3/priorities/{priority_id}"
)
if assignee_id:
payload["_links"]["assignee"] = build_link(f"/api/v3/users/{assignee_id}")
params = {"notify": str(notify).lower()}
result = await client.post(
f"projects/{project_id}/work_packages", data=payload
)
return result
finally:
await client.close()
def _format_work_package_markdown(wp: dict[str, Any]) -> str:
"""Format a work package as markdown.
Args:
wp: Work package object from OpenProject API
Returns:
Formatted markdown string
"""
# Basic information
wp_id = wp.get("id", "N/A")
subject = wp.get("subject", "No subject")
lock_version = wp.get("lockVersion", "N/A")
created_at = wp.get("createdAt", "N/A")
updated_at = wp.get("updatedAt", "N/A")
# Description
description_obj = wp.get("description", {})
description = description_obj.get("raw", "") if description_obj else ""
# Dates
start_date = wp.get("startDate", "Not set")
due_date = wp.get("dueDate", "Not set")
estimated_time = wp.get("estimatedTime", "Not set")
# Costs
labor_costs = wp.get("laborCosts", "0.00")
material_costs = wp.get("materialCosts", "0.00")
overall_costs = wp.get("overallCosts", "0.00")
spent_time = wp.get("spentTime", "PT0S")
# Get embedded resources
embedded = wp.get("_embedded", {})
# Type
type_obj = embedded.get("type", {})
type_name = type_obj.get("name", "N/A")
type_color = type_obj.get("color", "")
# Status
status_obj = embedded.get("status", {})
status_name = status_obj.get("name", "N/A")
status_color = status_obj.get("color", "")
# Priority
priority_obj = embedded.get("priority", {})
priority_name = priority_obj.get("name", "N/A")
# Project
project_obj = embedded.get("project", {})
project_name = project_obj.get("name", "N/A")
project_id = project_obj.get("identifier", "N/A")
# Author
author_obj = embedded.get("author", {})
author_name = author_obj.get("name", "Unknown")
# Assignee (from _links)
links = wp.get("_links", {})
assignee_link = links.get("assignee", {})
assignee_name = assignee_link.get("title") if assignee_link.get("href") else "Unassigned"
# Parent
parent_link = links.get("parent", {})
parent_info = parent_link.get("title") if parent_link.get("href") else "None"
# Ancestors
ancestors = links.get("ancestors", [])
ancestor_info = []
for ancestor in ancestors:
ancestor_info.append(ancestor.get("title", "Unknown"))
# Children
children = links.get("children", [])
children_count = len(children) if children else 0
# Attachments
attachments_obj = embedded.get("attachments", {})
attachments_count = attachments_obj.get("total", 0)
# Relations
relations_obj = embedded.get("relations", {})
relations_count = relations_obj.get("total", 0)
# Build markdown
markdown = f"""# Work Package #{wp_id}: {subject}
## Basic Information
- **Type:** {type_name}
- **Status:** {status_name}
- **Priority:** {priority_name}
- **Project:** {project_name} (`{project_id}`)
- **Lock Version:** {lock_version}
## People
- **Author:** {author_name}
- **Assignee:** {assignee_name}
## Dates & Schedule
- **Start Date:** {start_date}
- **Due Date:** {due_date}
- **Estimated Time:** {estimated_time}
- **Spent Time:** {spent_time}
## Costs
- **Labor Costs:** {labor_costs}
- **Material Costs:** {material_costs}
- **Overall Costs:** {overall_costs}
## Hierarchy
- **Parent:** {parent_info}"""
if ancestor_info:
markdown += "\n- **Ancestors:** " + " > ".join(ancestor_info)
if children_count > 0:
markdown += f"\n- **Children:** {children_count} child work package(s)"
markdown += f"""
## Additional Info
- **Attachments:** {attachments_count}
- **Relations:** {relations_count}
- **Created:** {created_at}
- **Updated:** {updated_at}
"""
if description:
markdown += f"""
## Description
```
{description}
```
"""
return markdown
async def get_work_package(work_package_id: int) -> str:
"""Get detailed information about a work package.
Args:
work_package_id: Work package ID
Returns:
Formatted markdown string with work package details
"""
client = OpenProjectClient()
try:
result = await client.get(f"work_packages/{work_package_id}")
return _format_work_package_markdown(result)
finally:
await client.close()
async def update_work_package(
work_package_id: int,
lock_version: int,
subject: str | None = None,
description: str | None = None,
status_id: int | None = None,
priority_id: int | None = None,
assignee_id: int | None = None,
notify: bool = True,
) -> dict[str, Any]:
"""Update an existing work package.
Args:
work_package_id: Work package ID
lock_version: Current lock version (get from work package first)
subject: New subject/title
description: New description in markdown
status_id: New status ID
priority_id: New priority ID
assignee_id: New assignee user ID (use 0 to unassign)
notify: Whether to send notifications (default: True)
Returns:
Updated work package object
"""
client = OpenProjectClient()
try:
payload: dict[str, Any] = {
"lockVersion": lock_version,
"_links": {},
}
if subject is not None:
payload["subject"] = subject
if description is not None:
payload["description"] = build_formattable(description)
if status_id is not None:
payload["_links"]["status"] = build_link(f"/api/v3/statuses/{status_id}")
if priority_id is not None:
payload["_links"]["priority"] = build_link(
f"/api/v3/priorities/{priority_id}"
)
if assignee_id is not None:
if assignee_id == 0:
payload["_links"]["assignee"] = None
else:
payload["_links"]["assignee"] = build_link(
f"/api/v3/users/{assignee_id}"
)
params = {"notify": str(notify).lower()}
result = await client.patch(
f"work_packages/{work_package_id}?notify={params['notify']}", data=payload
)
return result
finally:
await client.close()
def _format_work_package_list_item(wp: dict[str, Any]) -> str:
"""Format a work package as a concise list item.
Args:
wp: Work package object from OpenProject API
Returns:
Formatted markdown string for list display
"""
wp_id = wp.get("id", "N/A")
subject = wp.get("subject", "No subject")
# Get embedded resources
embedded = wp.get("_embedded", {})
links = wp.get("_links", {})
# Type
type_obj = embedded.get("type", {})
type_name = type_obj.get("name", "N/A")
# Status
status_obj = embedded.get("status", {})
status_name = status_obj.get("name", "N/A")
# Priority
priority_obj = embedded.get("priority", {})
priority_name = priority_obj.get("name", "N/A")
# Project
project_obj = embedded.get("project", {})
project_name = project_obj.get("name", "N/A")
# Assignee
assignee_link = links.get("assignee", {})
assignee_name = assignee_link.get("title") if assignee_link.get("href") else "Unassigned"
# Parent
parent_link = links.get("parent", {})
parent_name = parent_link.get("title") if parent_link.get("href") else None
# Due date
due_date = wp.get("dueDate", "No due date")
markdown = f"""### #{wp_id}: {subject}
- **Type:** {type_name} | **Status:** {status_name} | **Priority:** {priority_name}
- **Project:** {project_name} | **Assignee:** {assignee_name}
- **Due Date:** {due_date}"""
if parent_name:
markdown += f" | **Parent:** {parent_name}"
return markdown
async def list_work_packages(
project_id: str | None = None,
filters: str | None = None,
page: int = 1,
page_size: int = 20,
) -> str:
"""List work packages with optional filtering and pagination.
Args:
project_id: Optional project ID to filter work packages
filters: Optional JSON filter string (e.g., '[{"status_id":{"operator":"=","values":["1"]}}]')
page: Page number (default: 1)
page_size: Items per page (default: 20)
Returns:
Formatted markdown string with work packages list
"""
client = OpenProjectClient()
try:
params: dict[str, Any] = {
"pageSize": page_size,
"offset": (page - 1) * page_size,
}
if filters:
params["filters"] = filters
if project_id:
endpoint = f"projects/{project_id}/work_packages"
title = f"Work Packages in Project '{project_id}'"
else:
endpoint = "work_packages"
title = "All Work Packages"
result = await client.get(endpoint, params=params)
# Extract metadata
total = result.get("total", 0)
count = result.get("count", 0)
# Get work packages
work_packages = get_embedded_collection(result, "elements")
# Format as markdown
markdown = f"""# {title}
**Total:** {total} | **Showing:** {count} | **Page:** {page}
---
"""
if not work_packages:
markdown += "\n*No work packages found.*\n"
else:
for wp in work_packages:
markdown += "\n" + _format_work_package_list_item(wp) + "\n"
return markdown
finally:
await client.close()
async def delete_work_package(work_package_id: int) -> dict[str, Any]:
"""Permanently delete a work package.
Args:
work_package_id: Work package ID to delete
Returns:
Success confirmation
"""
client = OpenProjectClient()
try:
result = await client.delete(f"work_packages/{work_package_id}")
return result
finally:
await client.close()
async def get_available_assignees(project_id: str) -> dict[str, Any]:
"""Get list of users who can be assigned to work packages in a project.
Args:
project_id: Project identifier or ID
Returns:
Collection of available assignee users
"""
client = OpenProjectClient()
try:
result = await client.get(f"projects/{project_id}/available_assignees")
return result
finally:
await client.close()
async def add_watcher(work_package_id: int, user_id: int) -> dict[str, Any]:
"""Add a user as a watcher to a work package.
Args:
work_package_id: Work package ID
user_id: User ID to add as watcher
Returns:
Updated work package with watchers information
"""
client = OpenProjectClient()
try:
payload = {"user": build_link(f"/api/v3/users/{user_id}")}
result = await client.post(
f"work_packages/{work_package_id}/watchers", data=payload
)
return result
finally:
await client.close()
async def remove_watcher(work_package_id: int, user_id: int) -> dict[str, Any]:
"""Remove a user from work package watchers.
Args:
work_package_id: Work package ID
user_id: User ID to remove from watchers
Returns:
Success confirmation
"""
client = OpenProjectClient()
try:
result = await client.delete(
f"work_packages/{work_package_id}/watchers/{user_id}"
)
return result
finally:
await client.close()
async def get_work_package_schema(
project_id: str, type_id: int | None = None
) -> dict[str, Any]:
"""Get the schema for creating/updating work packages.
This returns available fields, required fields, and allowed values
for a work package in a specific project.
Args:
project_id: Project identifier or ID
type_id: Optional type ID to get type-specific schema
Returns:
Schema object with available fields and constraints
"""
client = OpenProjectClient()
try:
params = {}
if type_id:
params["filters"] = f'[{{"type":{{"operator":"=","values":["{type_id}"]}}}}]'
result = await client.get(
f"projects/{project_id}/work_packages/schema", params=params
)
return result
finally:
await client.close()
async def set_parent_work_package(
work_package_id: int,
parent_id: int | None,
lock_version: int,
notify: bool = True,
) -> dict[str, Any]:
"""Set or remove the parent of a work package.
Args:
work_package_id: Child work package ID
parent_id: Parent work package ID (use None to remove parent)
lock_version: Current lock version (get from work package first)
notify: Whether to send notifications (default: True)
Returns:
Updated work package object
"""
client = OpenProjectClient()
try:
payload: dict[str, Any] = {
"lockVersion": lock_version,
"_links": {},
}
if parent_id is None:
# Remove parent by setting it to null
payload["_links"]["parent"] = None
else:
# Set parent
payload["_links"]["parent"] = build_link(
f"/api/v3/work_packages/{parent_id}"
)
params = {"notify": str(notify).lower()}
result = await client.patch(
f"work_packages/{work_package_id}?notify={params['notify']}", data=payload
)
return result
finally:
await client.close()
