diff --git a/plane_mcp/tools/__init__.py b/plane_mcp/tools/__init__.py index ea2296b..6e01903 100644 --- a/plane_mcp/tools/__init__.py +++ b/plane_mcp/tools/__init__.py @@ -3,6 +3,7 @@ from fastmcp import FastMCP from plane_mcp.tools.cycles import register_cycle_tools +from plane_mcp.tools.epics import register_epic_tools from plane_mcp.tools.initiatives import register_initiative_tools from plane_mcp.tools.intake import register_intake_tools from plane_mcp.tools.labels import register_label_tools @@ -32,6 +33,7 @@ def register_tools(mcp: FastMCP) -> None: register_work_item_relation_tools(mcp) register_work_log_tools(mcp) register_cycle_tools(mcp) + register_epic_tools(mcp) register_user_tools(mcp) register_module_tools(mcp) register_initiative_tools(mcp) diff --git a/plane_mcp/tools/epics.py b/plane_mcp/tools/epics.py new file mode 100644 index 0000000..4c81b91 --- /dev/null +++ b/plane_mcp/tools/epics.py @@ -0,0 +1,296 @@ +"""Epic-related tools for Plane MCP Server.""" + +from typing import get_args + +from fastmcp import FastMCP +from plane import PlaneClient +from plane.models.enums import PriorityEnum +from plane.models.epics import Epic, PaginatedEpicResponse +from plane.models.query_params import PaginatedQueryParams, RetrieveQueryParams +from plane.models.work_item_types import WorkItemType +from plane.models.work_items import CreateWorkItem, UpdateWorkItem + +from plane_mcp.client import get_plane_client_context + + +def _get_epic_work_item_type( + client: PlaneClient, workspace_slug: str, project_id: str +) -> WorkItemType | None: + """Find the work item type marked as epic for a project.""" + response = client.work_item_types.list( + workspace_slug=workspace_slug, + project_id=project_id, + ) + for work_item_type in response: + if work_item_type.is_epic: + return work_item_type + return None + + +def register_epic_tools(mcp: FastMCP) -> None: + """Register all epic-related tools with the MCP server.""" + + @mcp.tool() + def list_epics( + project_id: str, + cursor: str | None = None, + per_page: int | None = None, + expand: str | None = None, + fields: str | None = None, + order_by: str | None = None, + ) -> list[Epic]: + """ + List all epics in a project. + + Args: + project_id: UUID of the project + cursor: Pagination cursor for getting next set of results + per_page: Number of results per page (1-100) + expand: Comma-separated list of related fields to expand in response + fields: Comma-separated list of fields to include in response + order_by: Field to order results by. Prefix with '-' for descending order + + Returns: + List of Epic objects + """ + client, workspace_slug = get_plane_client_context() + + params = PaginatedQueryParams( + cursor=cursor, + per_page=per_page, + expand=expand, + fields=fields, + order_by=order_by, + ) + + response: PaginatedEpicResponse = client.epics.list( + workspace_slug=workspace_slug, + project_id=project_id, + params=params, + ) + + return response.results + + @mcp.tool() + def create_epic( + project_id: str, + name: str, + assignees: list[str] | None = None, + labels: list[str] | None = None, + point: int | None = None, + description_html: str | None = None, + description_stripped: str | None = None, + priority: str | None = None, + start_date: str | None = None, + target_date: str | None = None, + sort_order: float | None = None, + is_draft: bool | None = None, + external_source: str | None = None, + external_id: str | None = None, + parent: str | None = None, + state: str | None = None, + estimate_point: str | None = None, + ) -> Epic: + """ + Create a new epic. + + Epics are work items with the epic type. This tool automatically resolves + the epic work item type for the project and creates the work item with it. + + Args: + project_id: UUID of the project + name: Epic name (required) + assignees: List of user IDs to assign to the epic + labels: List of label IDs to attach to the epic + point: Story point value + description_html: HTML description of the epic + description_stripped: Plain text description (stripped of HTML) + priority: Priority level (urgent, high, medium, low, none) + start_date: Start date (ISO 8601 format) + target_date: Target/end date (ISO 8601 format) + sort_order: Sort order value + is_draft: Whether the epic is a draft + external_source: External system source name + external_id: External system identifier + parent: UUID of the parent epic + state: UUID of the state + estimate_point: Estimate point value + + Returns: + Created Epic object + """ + client, workspace_slug = get_plane_client_context() + + epic_type = _get_epic_work_item_type(client, workspace_slug, project_id) + if epic_type is None: + raise ValueError("No work item type with is_epic=True found in the project") + + valid_priorities = get_args(PriorityEnum) + if priority is not None and priority not in valid_priorities: + raise ValueError(f"Invalid priority '{priority}'. Must be one of: {valid_priorities}") + validated_priority: PriorityEnum | None = priority # type: ignore[assignment] + + data = CreateWorkItem( + name=name, + assignees=assignees, + labels=labels, + type_id=epic_type.id, + point=point, + description_html=description_html, + description_stripped=description_stripped, + priority=validated_priority, + start_date=start_date, + target_date=target_date, + sort_order=sort_order, + is_draft=is_draft, + external_source=external_source, + external_id=external_id, + parent=parent, + state=state, + estimate_point=estimate_point, + ) + + work_item = client.work_items.create( + workspace_slug=workspace_slug, project_id=project_id, data=data + ) + + return client.epics.retrieve( + workspace_slug=workspace_slug, + project_id=project_id, + epic_id=work_item.id, + ) + + @mcp.tool() + def retrieve_epic( + project_id: str, + epic_id: str, + expand: str | None = None, + fields: str | None = None, + ) -> Epic: + """ + Retrieve an epic by ID. + + Args: + project_id: UUID of the project + epic_id: UUID of the epic + expand: Comma-separated list of related fields to expand in response + fields: Comma-separated list of fields to include in response + + Returns: + Epic object + """ + client, workspace_slug = get_plane_client_context() + + params = RetrieveQueryParams( + expand=expand, + fields=fields, + ) + + return client.epics.retrieve( + workspace_slug=workspace_slug, + project_id=project_id, + epic_id=epic_id, + params=params, + ) + + @mcp.tool() + def update_epic( + project_id: str, + epic_id: str, + name: str | None = None, + assignees: list[str] | None = None, + labels: list[str] | None = None, + point: int | None = None, + description_html: str | None = None, + description_stripped: str | None = None, + priority: str | None = None, + start_date: str | None = None, + target_date: str | None = None, + sort_order: float | None = None, + is_draft: bool | None = None, + external_source: str | None = None, + external_id: str | None = None, + state: str | None = None, + estimate_point: str | None = None, + ) -> Epic: + """ + Update an epic by ID. + + Args: + project_id: UUID of the project + epic_id: UUID of the epic + name: Epic name + assignees: List of user IDs to assign to the epic + labels: List of label IDs to attach to the epic + point: Story point value + description_html: HTML description of the epic + description_stripped: Plain text description (stripped of HTML) + priority: Priority level (urgent, high, medium, low, none) + start_date: Start date (ISO 8601 format) + target_date: Target/end date (ISO 8601 format) + sort_order: Sort order value + is_draft: Whether the epic is a draft + external_source: External system source name + external_id: External system identifier + state: UUID of the state + estimate_point: Estimate point value + + Returns: + Updated Epic object + """ + client, workspace_slug = get_plane_client_context() + + valid_priorities = get_args(PriorityEnum) + if priority is not None and priority not in valid_priorities: + raise ValueError(f"Invalid priority '{priority}'. Must be one of: {valid_priorities}") + validated_priority: PriorityEnum | None = priority # type: ignore[assignment] + + data = UpdateWorkItem( + name=name, + assignees=assignees, + labels=labels, + point=point, + description_html=description_html, + description_stripped=description_stripped, + priority=validated_priority, + start_date=start_date, + target_date=target_date, + sort_order=sort_order, + is_draft=is_draft, + external_source=external_source, + external_id=external_id, + state=state, + estimate_point=estimate_point, + ) + + work_item = client.work_items.update( + workspace_slug=workspace_slug, + project_id=project_id, + work_item_id=epic_id, + data=data, + ) + + return client.epics.retrieve( + workspace_slug=workspace_slug, + project_id=project_id, + epic_id=work_item.id, + ) + + @mcp.tool() + def delete_epic( + project_id: str, + epic_id: str, + ) -> None: + """ + Delete an epic by ID. + + Args: + project_id: UUID of the project + epic_id: UUID of the epic + """ + client, workspace_slug = get_plane_client_context() + client.work_items.delete( + workspace_slug=workspace_slug, + project_id=project_id, + work_item_id=epic_id, + ) diff --git a/tests/test_integration.py b/tests/test_integration.py index 2a49064..6a53f22 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -56,8 +56,12 @@ async def run_integration_test(): 2. Create work item 1 3. Create work item 2 4. Update work item 2 with work item 1 as parent - 5. Delete work items - 6. Delete project + 5. Create epic + 6. Update work item 2 to be under the epic + 7. List epics in project + 8. Delete epic + 9. Delete work items + 10. Delete project """ config = get_config() unique_id = uuid.uuid4().hex[:6] @@ -123,7 +127,53 @@ async def run_integration_test(): ) print(f"Set work item 1 as parent of work item 2") - # 5. Delete work items + # 5. Create epic + print("Creating epic...") + epic_result = await client.call_tool( + "create_epic", + { + "project_id": project_id, + "name": f"Epic {unique_id}", + }, + ) + epic = extract_result(epic_result) + epic_id = epic["id"] + print(f"Created epic: {epic_id}") + + # 6. Update work item 2 to be under the epic + print("Setting epic as parent of work item 2...") + await client.call_tool( + "update_work_item", + { + "project_id": project_id, + "work_item_id": work_item_2_id, + "parent": epic_id, + }, + ) + print("Set epic as parent of work item 2") + + # 7. List epics in project + print("Listing epics in project...") + epics_result = await client.call_tool( + "list_epics", + { + "project_id": project_id, + }, + ) + epics = extract_result(epics_result) + epic_ids = {e["id"] for e in epics} + print(f"Epics in project: {list(epic_ids)}") + assert epic_id in epic_ids, "Created epic not found in list_epics results" + + # 8. Delete epic + print("Deleting epic...") + await client.call_tool( + "delete_epic", + {"project_id": project_id, "epic_id": epic_id}, + ) + print("Deleted epic") + + # 9. Delete work items print(f"Deleting work items...") await client.call_tool( "delete_work_item", @@ -137,7 +187,7 @@ async def run_integration_test(): ) print(f"Deleted work item 1") - # 6. Delete project + # 10. Delete project print(f"Deleting project...") await client.call_tool("delete_project", {"project_id": project_id}) print(f"Deleted project") @@ -246,6 +296,12 @@ def test_full_integration(): "retrieve_initiative", "update_initiative", "delete_initiative", + # Epic tools + "list_epics", + "create_epic", + "retrieve_epic", + "update_epic", + "delete_epic", # Intake tools "list_intake_work_items", "create_intake_work_item",