"""Main MCP server for OpenProject API v3 integration."""
from mcp.server.fastmcp import FastMCP
from .tools import comments, projects, relations, work_packages
# Initialize the MCP server
mcp = FastMCP(
"openproject",
)
# Work Package Tools
@mcp.tool()
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,
):
"""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)
"""
return await work_packages.create_work_package(
project_id=project_id,
subject=subject,
description=description,
type_id=type_id,
status_id=status_id,
priority_id=priority_id,
assignee_id=assignee_id,
notify=notify,
)
@mcp.tool()
async def get_work_package(work_package_id: int):
"""Get detailed information about a work package.
Args:
work_package_id: Work package ID
"""
return await work_packages.get_work_package(work_package_id=work_package_id)
@mcp.tool()
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,
):
"""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)
"""
return await work_packages.update_work_package(
work_package_id=work_package_id,
lock_version=lock_version,
subject=subject,
description=description,
status_id=status_id,
priority_id=priority_id,
assignee_id=assignee_id,
notify=notify,
)
@mcp.tool()
async def list_work_packages(
project_id: str | None = None,
filters: str | None = None,
page: int = 1,
page_size: int = 20,
):
"""List work packages with optional filtering and pagination.
Args:
project_id: Optional project ID to filter work packages
filters: Optional JSON filter string
page: Page number (default: 1)
page_size: Items per page (default: 20)
"""
return await work_packages.list_work_packages(
project_id=project_id,
filters=filters,
page=page,
page_size=page_size,
)
@mcp.tool()
async def delete_work_package(work_package_id: int):
"""Permanently delete a work package.
Args:
work_package_id: Work package ID to delete
"""
return await work_packages.delete_work_package(work_package_id=work_package_id)
@mcp.tool()
async def get_available_assignees(project_id: str):
"""Get list of users who can be assigned to work packages in a project.
Args:
project_id: Project identifier or ID
"""
return await work_packages.get_available_assignees(project_id=project_id)
@mcp.tool()
async def add_watcher(work_package_id: int, user_id: int):
"""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
"""
return await work_packages.add_watcher(
work_package_id=work_package_id, user_id=user_id
)
@mcp.tool()
async def remove_watcher(work_package_id: int, user_id: int):
"""Remove a user from work package watchers.
Args:
work_package_id: Work package ID
user_id: User ID to remove from watchers
"""
return await work_packages.remove_watcher(
work_package_id=work_package_id, user_id=user_id
)
@mcp.tool()
async def get_work_package_schema(project_id: str, type_id: int | None = None):
"""Get the schema for creating/updating work packages.
Args:
project_id: Project identifier or ID
type_id: Optional type ID to get type-specific schema
"""
return await work_packages.get_work_package_schema(
project_id=project_id, type_id=type_id
)
@mcp.tool()
async def set_parent_work_package(
work_package_id: int,
parent_id: int | None,
lock_version: int,
notify: bool = True,
):
"""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)
"""
return await work_packages.set_parent_work_package(
work_package_id=work_package_id,
parent_id=parent_id,
lock_version=lock_version,
notify=notify,
)
# Comment/Activity Tools
@mcp.tool()
async def get_work_package_activities(
work_package_id: int, page: int = 1, page_size: int = 20
):
"""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)
"""
return await comments.get_work_package_activities(
work_package_id=work_package_id,
page=page,
page_size=page_size,
)
@mcp.tool()
async def create_comment(work_package_id: int, comment: str, internal: bool = False):
"""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)
"""
return await comments.create_comment(
work_package_id=work_package_id,
comment=comment,
internal=internal,
)
@mcp.tool()
async def get_activity(activity_id: int):
"""Get details of a specific activity.
Args:
activity_id: Activity ID
"""
return await comments.get_activity(activity_id=activity_id)
@mcp.tool()
async def update_comment(activity_id: int, comment: str, lock_version: int):
"""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)
"""
return await comments.update_comment(
activity_id=activity_id,
comment=comment,
lock_version=lock_version,
)
# Project Tools
@mcp.tool()
async def get_project(project_id: str):
"""Get detailed information about a project.
Args:
project_id: Project identifier or ID
"""
return await projects.get_project(project_id=project_id)
@mcp.tool()
async def list_projects(
filters: str | None = None, page: int = 1, page_size: int = 20
):
"""List all accessible projects with optional filtering and pagination.
Args:
filters: Optional JSON filter string
page: Page number (default: 1)
page_size: Items per page (default: 20)
"""
return await projects.list_projects(
filters=filters,
page=page,
page_size=page_size,
)
@mcp.tool()
async def update_project(
project_id: str,
lock_version: int,
name: str | None = None,
description: str | None = None,
public: bool | None = None,
active: bool | None = None,
):
"""Update an existing project.
Args:
project_id: Project identifier or ID
lock_version: Current lock version (get from project first)
name: New project name
description: New project description in markdown
public: Whether project is public
active: Whether project is active
"""
return await projects.update_project(
project_id=project_id,
lock_version=lock_version,
name=name,
description=description,
public=public,
active=active,
)
# Relation Tools
@mcp.tool()
async def list_work_package_relations(work_package_id: int):
"""Get all relations for a work package.
Args:
work_package_id: Work package ID
"""
return await relations.list_work_package_relations(
work_package_id=work_package_id
)
@mcp.tool()
async def create_relation(
from_id: int,
to_id: int,
relation_type: str,
lag: int | None = None,
):
"""Create a relation between two work packages.
Args:
from_id: Source work package ID
to_id: Target work package ID
relation_type: Type of relation (relates, duplicates, blocks, precedes, follows, includes, partof, requires)
lag: Optional lag in days for precedes/follows relations
"""
return await relations.create_relation(
from_id=from_id,
to_id=to_id,
relation_type=relation_type, # type: ignore
lag=lag,
)
@mcp.tool()
async def get_relation(relation_id: int):
"""Get details of a specific relation.
Args:
relation_id: Relation ID
"""
return await relations.get_relation(relation_id=relation_id)
@mcp.tool()
async def delete_relation(relation_id: int):
"""Remove a relation between work packages.
Args:
relation_id: Relation ID to delete
"""
return await relations.delete_relation(relation_id=relation_id)
def main():
"""Run the MCP server with stdio transport."""
mcp.run(transport="stdio")
if __name__ == "__main__":
main()
