Skip to content

feat: add work item attachment tools #119

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
volodchenkov wants to merge 1 commit into
base: main
Choose a base branch
from
+281 −21
Open

feat: add work item attachment tools #119

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions plane_mcp/tools/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)
Expand Down
200 changes: 200 additions & 0 deletions plane_mcp/tools/work_item_attachments.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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:
Comment thread
coderabbitai[bot] marked this conversation as resolved.
"""
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,
)
100 changes: 79 additions & 21 deletions tests/test_integration.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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,
},
)
Comment on lines +132 to +160
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot May 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Actually upload the file before flipping is_uploaded=True.

Lines 144-160 never use attachment_wrapper["upload_data"], so this test still passes if the presigned form policy or upload endpoint is broken. That misses the main failure mode in the new two-step attachment flow; POST a tiny multipart body to upload_data["url"] with the returned fields before calling update_work_item_attachment.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/test_integration.py` around lines 132 - 160, The test creates an
attachment via create_work_item_attachment but never actually uploads the file
bytes using attachment_wrapper["upload_data"], so the presigned/form upload step
is skipped; modify the test to POST a multipart/form-data to
attachment_wrapper["upload_data"]["url"] using the returned fields
(attachment_wrapper["upload_data"]["fields"] or similar) and include the file
body (e.g., 12 bytes or the expected content) before calling
update_work_item_attachment to set is_uploaded=True, keeping the existing
client.call_tool calls (create_work_item_attachment and
update_work_item_attachment) and using an async HTTP client compatible with the
test framework to perform the upload.

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...")

Expand Down Expand Up @@ -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],
},
)
Expand All @@ -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")

Expand Down Expand Up @@ -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",
Expand Down