feat: add epic CRUD tools

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)

plane_mcp/tools/epics.py

@@ -0,0 +1,295 @@
+"""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")
+
+        validated_priority: PriorityEnum | None = (
+            priority if priority in get_args(PriorityEnum) else None  # 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,
+        )