diff --git a/plane_mcp/tools/__init__.py b/plane_mcp/tools/__init__.py index 62bc43c..959c162 100644 --- a/plane_mcp/tools/__init__.py +++ b/plane_mcp/tools/__init__.py @@ -14,6 +14,7 @@ from plane_mcp.tools.states import register_state_tools from plane_mcp.tools.users import register_user_tools from plane_mcp.tools.work_item_activities import register_work_item_activity_tools +from plane_mcp.tools.work_item_attachments import register_work_item_attachment_tools from plane_mcp.tools.work_item_comments import register_work_item_comment_tools from plane_mcp.tools.work_item_links import register_work_item_link_tools from plane_mcp.tools.work_item_properties import register_work_item_property_tools @@ -29,6 +30,7 @@ def register_tools(mcp: FastMCP) -> None: register_project_tools(mcp) register_work_item_tools(mcp) register_work_item_activity_tools(mcp) + register_work_item_attachment_tools(mcp) register_work_item_comment_tools(mcp) register_work_item_link_tools(mcp) register_work_item_relation_tools(mcp) diff --git a/plane_mcp/tools/work_item_attachments.py b/plane_mcp/tools/work_item_attachments.py new file mode 100644 index 0000000..b189378 --- /dev/null +++ b/plane_mcp/tools/work_item_attachments.py @@ -0,0 +1,200 @@ +"""Work item attachment-related tools for Plane MCP Server.""" + +from typing import Any + +from fastmcp import FastMCP +from plane.models.work_items import ( + UpdateWorkItemAttachment, + WorkItemAttachment, + WorkItemAttachmentUploadRequest, +) +from pydantic import BaseModel, ConfigDict + +from plane_mcp.client import get_plane_client_context + + +class WorkItemAttachmentCreated(BaseModel): + """Wrapper response from create_work_item_attachment. + + Plane's POST .../issue-attachments/ returns a wrapper containing both the + attachment record and the S3 multipart-POST policy needed to upload the + file bytes. The caller posts the file as multipart/form-data to + ``upload_data["url"]`` with the ``upload_data["fields"]`` plus a ``file`` + part, then calls ``update_work_item_attachment`` with ``is_uploaded=True``. + """ + + model_config = ConfigDict(extra="allow") + + attachment: WorkItemAttachment + upload_data: dict[str, Any] + asset_id: str | None = None + asset_url: str | None = None + + +def register_work_item_attachment_tools(mcp: FastMCP) -> None: + """Register all work item attachment-related tools with the MCP server.""" + + @mcp.tool() + def list_work_item_attachments( + project_id: str, + work_item_id: str, + params: dict[str, Any] | None = None, + ) -> list[WorkItemAttachment]: + """ + List attachments for a work item. + + Args: + project_id: UUID of the project + work_item_id: UUID of the work item + params: Optional query parameters as a dictionary + + Returns: + List of WorkItemAttachment objects + """ + client, workspace_slug = get_plane_client_context() + return client.work_items.attachments.list( + workspace_slug=workspace_slug, + project_id=project_id, + work_item_id=work_item_id, + params=params, + ) + + @mcp.tool() + def create_work_item_attachment( + project_id: str, + work_item_id: str, + name: str, + size: int, + mime_type: str | None = None, + external_id: str | None = None, + external_source: str | None = None, + ) -> WorkItemAttachmentCreated: + """ + Register an attachment for a work item and get a presigned upload URL. + + Plane attachments use a two-step asset flow. This tool creates the + attachment record on the server and returns: + + * ``attachment`` — the created ``WorkItemAttachment`` record (with + ``id``, ``asset`` storage path, ``is_uploaded=False``, etc.) + * ``upload_data`` — an S3 multipart-POST policy. The caller posts the + file as ``multipart/form-data`` to ``upload_data["url"]`` with the + ``upload_data["fields"]`` plus a ``file`` part. + * ``asset_id`` and ``asset_url`` — convenience identifiers. + + After the upload completes, call ``update_work_item_attachment`` with + ``is_uploaded=True`` to mark the attachment ready. + + Note: this tool calls the underlying ``_post`` directly because the + plane-sdk ``WorkItemAttachments.create()`` method incorrectly + validates the wrapper response as ``WorkItemAttachment``. See + plane-python-sdk for the upstream fix. + + Args: + project_id: UUID of the project + work_item_id: UUID of the work item + name: Original filename of the asset + size: File size in bytes + mime_type: MIME type of the file + external_id: External identifier for the asset + external_source: External source system + + Returns: + WorkItemAttachmentCreated wrapper with ``attachment`` record and + ``upload_data`` S3 policy. + """ + client, workspace_slug = get_plane_client_context() + + data = WorkItemAttachmentUploadRequest( + name=name, + size=size, + type=mime_type, + external_id=external_id, + external_source=external_source, + ) + + raw = client.work_items.attachments._post( + f"{workspace_slug}/projects/{project_id}/work-items/{work_item_id}/attachments", + data.model_dump(exclude_none=True), + ) + return WorkItemAttachmentCreated.model_validate(raw) + + @mcp.tool() + def update_work_item_attachment( + project_id: str, + work_item_id: str, + attachment_id: str, + is_uploaded: bool, + ) -> WorkItemAttachment: + """ + Update an attachment for a work item. + + Typically used to confirm a successful binary upload by setting + ``is_uploaded=True`` after the caller has POSTed the file bytes to + the presigned URL returned by ``create_work_item_attachment``. + + Note: Plane responds to attachment PATCH with ``204 No Content`` and + exposes no metadata-by-id endpoint (the GET on a single attachment + URL is a download redirect). To return the updated record this tool + follows the PATCH with a list call and filters by id, mirroring + plane-python-sdk PR #34. Once that PR lands and the SDK pin is + bumped here, this can switch to calling + ``client.work_items.attachments.update`` directly. + + Args: + project_id: UUID of the project + work_item_id: UUID of the work item + attachment_id: UUID of the attachment + is_uploaded: Mark attachment as uploaded + + Returns: + Updated WorkItemAttachment object. + """ + if is_uploaded is not True: + raise ValueError( + "Only is_uploaded=True is currently supported (Plane lists/exposes only uploaded attachments)." + ) + + client, workspace_slug = get_plane_client_context() + + data = UpdateWorkItemAttachment(is_uploaded=True) + + client.work_items.attachments._patch( + f"{workspace_slug}/projects/{project_id}/work-items/{work_item_id}/attachments/{attachment_id}", + data.model_dump(exclude_none=True), + ) + for attachment in client.work_items.attachments.list( + workspace_slug=workspace_slug, + project_id=project_id, + work_item_id=work_item_id, + ): + if attachment.id == attachment_id: + return attachment + raise ValueError( + f"Attachment {attachment_id} not found after update; Plane only lists attachments with is_uploaded=True." + ) + + @mcp.tool() + def delete_work_item_attachment( + project_id: str, + work_item_id: str, + attachment_id: str, + ) -> None: + """ + Delete an attachment from a work item. + + Args: + project_id: UUID of the project + work_item_id: UUID of the work item + attachment_id: UUID of the attachment + + Returns: + None on success. + """ + client, workspace_slug = get_plane_client_context() + client.work_items.attachments.delete( + workspace_slug=workspace_slug, + project_id=project_id, + work_item_id=work_item_id, + attachment_id=attachment_id, + ) diff --git a/tests/test_integration.py b/tests/test_integration.py index 9ff5635..bb4ea5c 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -22,9 +22,7 @@ def get_config(): mcp_url = os.getenv("PLANE_TEST_MCP_URL", "http://localhost:8211") if not api_key or not workspace_slug: - raise RuntimeError( - "Missing required env vars: PLANE_TEST_API_KEY, PLANE_TEST_WORKSPACE_SLUG" - ) + raise RuntimeError("Missing required env vars: PLANE_TEST_API_KEY, PLANE_TEST_WORKSPACE_SLUG") return { "api_key": api_key, @@ -54,20 +52,20 @@ async def run_integration_test(): Full integration test: 1. Create a project 2. Create work item 1 - 3. Create work item 2 - 4. Update work item 2 with work item 1 as parent - 5. Create epic with work item 1 as the underlying work item - 6. Update work item 2 to be under the epic - 7. List all epics + 3. Create work item 2 + 4. Update work item 2 with work item 1 as parent + 5. Create epic with work item 1 as the underlying work item + 6. Update work item 2 to be under the epic + 7. List all epics 8. Create a milestone and associate it with the project and work items 9. Update the milestone to change its name and description 10. List all milestones in the project 11. Delete the milestone 12. Delete the epic - 13. Delete work items - 14. Delete project - """ - config = get_config() + 13. Delete work items + 14. Delete project + """ + config = get_config() unique_id = uuid.uuid4().hex[:6] transport = StreamableHttpTransport( @@ -131,6 +129,61 @@ async def run_integration_test(): ) print("Set work item 1 as parent of work item 2") + # Work item attachment lifecycle (create metadata → list → retrieve → mark uploaded → delete) + print("Creating work item attachment...") + attachment_result = await client.call_tool( + "create_work_item_attachment", + { + "project_id": project_id, + "work_item_id": work_item_1_id, + "name": f"test-{unique_id}.txt", + "size": 12, + "mime_type": "text/plain", + }, + ) + attachment_wrapper = extract_result(attachment_result) + # create_work_item_attachment returns WorkItemAttachmentCreated wrapper: + # {attachment, upload_data, asset_id, asset_url} + attachment_id = attachment_wrapper["attachment"]["id"] + print(f"Created attachment: {attachment_id}") + + # Plane filters list/retrieve to is_uploaded=True only — mark first. + print("Marking attachment as uploaded...") + await client.call_tool( + "update_work_item_attachment", + { + "project_id": project_id, + "work_item_id": work_item_1_id, + "attachment_id": attachment_id, + "is_uploaded": True, + }, + ) + print("Marked attachment as uploaded") + + print("Listing work item attachments...") + attachments_list_result = await client.call_tool( + "list_work_item_attachments", + { + "project_id": project_id, + "work_item_id": work_item_1_id, + }, + ) + attachments_list = extract_result(attachments_list_result) + attachment_ids = [a["id"] for a in attachments_list if isinstance(a, dict) and "id" in a] + print(f"Attachments on work item 1: {attachment_ids}") + assert attachment_id in attachment_ids, "Created attachment was not returned by list_work_item_attachments" + + print("Deleting attachment...") + await client.call_tool( + "delete_work_item_attachment", + { + "project_id": project_id, + "work_item_id": work_item_1_id, + "attachment_id": attachment_id, + }, + ) + print("Deleted attachment") + # 5. Create epic with work item 1 as the underlying work item print("Creating epic...") @@ -178,7 +231,7 @@ async def run_integration_test(): { "project_id": project_id, "name": f"Milestone {unique_id}", - "description": "Integration test milestone", + "description": "Integration test milestone", "associated_work_item_ids": [epic_id, work_item_1_id, work_item_2_id], }, ) @@ -199,18 +252,18 @@ async def run_integration_test(): print(f"Work items associated with milestone: {[wi['id'] for wi in milestone_work_items]}") print(f"Created milestone: {milestone_id}") - + # 9. Update the milestone to change its name and description print("Updating milestone...") await client.call_tool( - "update_milestone", - { - "project_id": project_id, - "milestone_id": milestone_id, - "name": f"Updated Milestone {unique_id}", - "description": "Updated description for integration test milestone" + "update_milestone", + { + "project_id": project_id, + "milestone_id": milestone_id, + "name": f"Updated Milestone {unique_id}", + "description": "Updated description for integration test milestone", }, - ) + ) print("Updated milestone") @@ -283,6 +336,11 @@ def test_full_integration(): # Work item activity tools "list_work_item_activities", "retrieve_work_item_activity", + # Work item attachment tools + "list_work_item_attachments", + "create_work_item_attachment", + "update_work_item_attachment", + "delete_work_item_attachment", # Work item comment tools "list_work_item_comments", "retrieve_work_item_comment",