Add multimodal support for Linear

src/codegen/extensions/events/linear.py

@@ -1,10 +1,13 @@
 import logging
-from typing import Any, Callable, TypeVar
+import os
+import tempfile
+from typing import Any, Callable, Optional, TypeVar
 
 from pydantic import BaseModel
 
 from codegen.extensions.events.interface import EventHandlerManagerProtocol
-from codegen.extensions.linear.types import LinearEvent
+from codegen.extensions.linear.linear_client import LinearClient
+from codegen.extensions.linear.types import LinearAttachment, LinearEvent
 from codegen.shared.logging.get_logger import get_logger
 
 logger = get_logger(__name__)
@@ -18,6 +21,18 @@ class Linear(EventHandlerManagerProtocol):
     def __init__(self, app):
         self.app = app
         self.registered_handlers = {}
+        self._client = None
+
+    @property
+    def client(self) -> LinearClient:
+        """Get the Linear client instance."""
+        if not self._client:
+            access_token = os.environ.get("LINEAR_ACCESS_TOKEN")
+            if not access_token:
+                msg = "LINEAR_ACCESS_TOKEN environment variable is not set"
+                raise ValueError(msg)
+            self._client = LinearClient(access_token=access_token)
+        return self._client
 
     def unsubscribe_all_handlers(self):
         logger.info("[HANDLERS] Clearing all handlers")
@@ -44,6 +59,18 @@ def new_func(raw_event: dict):
 
                 # Parse event into LinearEvent type
                 event = LinearEvent.model_validate(raw_event)
+
+                # Check if this is an issue event and has attachments
+                if event_type == "Issue" and hasattr(event.data, "id"):
+                    try:
+                        # Get attachments for the issue
+                        attachments = self.client.get_issue_attachments(event.data.id)
+                        if attachments:
+                            event.attachments = attachments
+                            logger.info(f"[HANDLER] Found {len(attachments)} attachments for issue {event.data.id}")
+                    except Exception as e:
+                        logger.exception(f"[HANDLER] Error getting attachments: {e}")
+
                 return func(event)
 
             self.registered_handlers[event_name] = new_func
@@ -83,3 +110,48 @@ async def handle(self, event: dict) -> dict:
         except Exception as e:
             logger.exception(f"Error handling Linear event: {e}")
             return {"error": f"Failed to handle event: {e!s}"}
+
+    def download_attachment(self, attachment: LinearAttachment, directory: Optional[str] = None) -> str:
+        """Download a file attachment from Linear.
+
+        Args:
+            attachment: The LinearAttachment object
+            directory: Optional directory to save the file to. If not provided, uses a temporary directory.
+
+        Returns:
+            Path to the downloaded file
+        """
+        try:
+            # Download the attachment
+            content = self.client.download_attachment(attachment.url)
+
+            # Determine file path
+            if directory:
+                os.makedirs(directory, exist_ok=True)
+                file_path = os.path.join(directory, attachment.title)
+            else:
+                # Create a temporary file
+                temp_dir = tempfile.mkdtemp()
+                file_path = os.path.join(temp_dir, attachment.title)
+
+            # Write the file
+            with open(file_path, "wb") as f:
+                f.write(content)
+
+            logger.info(f"[HANDLER] Downloaded attachment to {file_path}")
+            return file_path
+        except Exception as e:
+            logger.exception(f"[HANDLER] Error downloading attachment: {e}")
+            msg = f"Failed to download attachment: {e}"
+            raise Exception(msg)
+
+    def upload_file(self, file_path: str) -> str:
+        """Upload a file to Linear.
+
+        Args:
+            file_path: Path to the file to upload
+
+        Returns:
+            URL of the uploaded file
+        """
+        return self.client.upload_file(file_path)

src/codegen/extensions/linear/linear_client.py

@@ -1,11 +1,13 @@
+import mimetypes
 import os
+from pathlib import Path
 from typing import Optional
 
 import requests
 from requests.adapters import HTTPAdapter
 from urllib3.util.retry import Retry
 
-from codegen.extensions.linear.types import LinearComment, LinearIssue, LinearTeam, LinearUser
+from codegen.extensions.linear.types import LinearAttachment, LinearComment, LinearIssue, LinearTeam, LinearUploadResponse, LinearUser
 from codegen.shared.logging.get_logger import get_logger
 
 logger = get_logger(__name__)
@@ -293,3 +295,171 @@ def get_teams(self) -> list[LinearTeam]:
         except Exception as e:
             msg = f"Error getting teams\n{data}\n{e}"
             raise Exception(msg)
+
+    def get_issue_attachments(self, issue_id: str) -> list[LinearAttachment]:
+        """Get all attachments for an issue.
+
+        Args:
+            issue_id: ID of the issue to get attachments for
+
+        Returns:
+            List of LinearAttachment objects
+        """
+        query = """
+            query getIssueAttachments($issueId: String!) {
+                issue(id: $issueId) {
+                    attachments {
+                        nodes {
+                            id
+                            url
+                            title
+                            subtitle
+                            size
+                            contentType
+                            source
+                        }
+                    }
+                }
+            }
+        """
+
+        variables = {"issueId": issue_id}
+        response = self.session.post(
+            self.api_endpoint,
+            headers=self.api_headers,
+            json={"query": query, "variables": variables},
+        )
+        data = response.json()
+
+        try:
+            attachments_data = data["data"]["issue"]["attachments"]["nodes"]
+            return [
+                LinearAttachment(
+                    id=attachment["id"],
+                    url=attachment["url"],
+                    title=attachment["title"],
+                    subtitle=attachment.get("subtitle"),
+                    size=attachment.get("size"),
+                    content_type=attachment.get("contentType"),
+                    source=attachment.get("source"),
+                    issue_id=issue_id,
+                )
+                for attachment in attachments_data
+            ]
+        except Exception as e:
+            logger.exception(f"Error getting issue attachments: {e}")
+            return []
+
+    def download_attachment(self, attachment_url: str) -> bytes:
+        """Download a file attachment from Linear.
+
+        Args:
+            attachment_url: URL of the attachment to download
+
+        Returns:
+            Binary content of the attachment
+        """
+        # Linear files are stored at uploads.linear.app
+        headers = {"Authorization": f"Bearer {self.access_token}"}
+
+        try:
+            response = self.session.get(attachment_url, headers=headers)
+            response.raise_for_status()
+            return response.content
+        except Exception as e:
+            logger.exception(f"Error downloading attachment: {e}")
+            msg = f"Failed to download attachment: {e}"
+            raise Exception(msg)
+
+    def request_upload_url(self, content_type: str, filename: str, size: int) -> LinearUploadResponse:
+        """Request a pre-signed URL for file upload.
+
+        Args:
+            content_type: MIME type of the file
+            filename: Name of the file
+            size: Size of the file in bytes
+
+        Returns:
+            LinearUploadResponse with upload URL and headers
+        """
+        mutation = """
+            mutation FileUpload($contentType: String!, $filename: String!, $size: Int!) {
+                fileUpload(contentType: $contentType, filename: $filename, size: $size) {
+                    success
+                    uploadFile {
+                        assetUrl
+                        uploadUrl
+                        headers {
+                            key
+                            value
+                        }
+                    }
+                }
+            }
+        """
+
+        variables = {
+            "contentType": content_type,
+            "filename": filename,
+            "size": size,
+        }
+
+        response = self.session.post(
+            self.api_endpoint,
+            headers=self.api_headers,
+            json={"query": mutation, "variables": variables},
+        )
+        data = response.json()
+
+        try:
+            return LinearUploadResponse.model_validate(data["data"])
+        except Exception as e:
+            logger.exception(f"Error requesting upload URL: {e}")
+            msg = f"Failed to request upload URL: {e}"
+            raise Exception(msg)
+
+    def upload_file(self, file_path: str) -> str:
+        """Upload a file to Linear.
+
+        Args:
+            file_path: Path to the file to upload
+
+        Returns:
+            URL of the uploaded file
+        """
+        path = Path(file_path)
+        if not path.exists():
+            msg = f"File not found: {file_path}"
+            raise FileNotFoundError(msg)
+
+        # Get file info
+        file_size = path.stat().st_size
+        file_name = path.name
+        content_type = mimetypes.guess_type(file_path)[0] or "application/octet-stream"
+
+        # Request upload URL
+        upload_response = self.request_upload_url(content_type, file_name, file_size)
+
+        if not upload_response.success or not upload_response.uploadFile:
+            msg = "Failed to request upload URL"
+            raise Exception(msg)
+
+        upload_url = upload_response.uploadFile.uploadUrl
+        asset_url = upload_response.uploadFile.assetUrl
+
+        # Prepare headers for the PUT request
+        headers = {
+            "Content-Type": content_type,
+            "Cache-Control": "public, max-age=31536000",
+        }
+
+        # Add headers from the upload response
+        for header in upload_response.uploadFile.headers:
+            headers[header.key] = header.value
+
+        # Upload the file
+        with open(file_path, "rb") as file:
+            response = self.session.put(upload_url, headers=headers, data=file)
+            response.raise_for_status()
+
+        return asset_url

src/codegen/extensions/linear/types.py

@@ -1,4 +1,6 @@
-from pydantic import BaseModel
+from typing import Optional
+
+from pydantic import BaseModel, Field
 
 
 class LinearUser(BaseModel):
@@ -28,6 +30,41 @@ class LinearIssue(BaseModel):
     team_id: str | None = None
 
 
+class LinearAttachment(BaseModel):
+    """Represents a file attachment in Linear."""
+
+    id: str
+    url: str
+    title: str
+    subtitle: Optional[str] = None
+    size: Optional[int] = None
+    content_type: Optional[str] = None
+    source: Optional[str] = None
+    issue_id: Optional[str] = None
+
+
+class LinearUploadHeader(BaseModel):
+    """Header for file upload."""
+
+    key: str
+    value: str
+
+
+class LinearUploadFile(BaseModel):
+    """Response from file upload request."""
+
+    assetUrl: str
+    uploadUrl: str
+    headers: list[LinearUploadHeader]
+
+
+class LinearUploadResponse(BaseModel):
+    """Response from fileUpload mutation."""
+
+    success: bool
+    uploadFile: LinearUploadFile
+
+
 class LinearEvent(BaseModel):
     """Represents a Linear webhook event."""
 
@@ -38,3 +75,4 @@ class LinearEvent(BaseModel):
     created_at: str | None = None  # ISO timestamp
     organization_id: str | None = None
     team_id: str | None = None
+    attachments: list[LinearAttachment] = Field(default_factory=list)  # Attachments associated with the event