feat: adds linear tools

src/codegen/extensions/langchain/tools.py

@@ -7,6 +7,12 @@
 from pydantic import BaseModel, Field
 
 from codegen import Codebase
+from codegen.extensions.linear.linear_client import LinearClient
+from codegen.extensions.tools.linear_tools import (
+    linear_comment_on_issue_tool,
+    linear_get_issue_comments_tool,
+    linear_get_issue_tool,
+)
 
 from ..tools import (
     commit,
@@ -37,9 +43,9 @@
 class ViewFileTool(BaseTool):
     """Tool for viewing file contents and metadata."""
 
     name: ClassVar[str] = "view_file"
     description: ClassVar[str] = "View the contents and metadata of a file in the codebase"
     args_schema: ClassVar[type[BaseModel]] = ViewFileInput
     codebase: Codebase = Field(exclude=True)
 
     def __init__(self, codebase: Codebase) -> None:
@@ -60,9 +66,9 @@
 class ListDirectoryTool(BaseTool):
     """Tool for listing directory contents."""
 
     name: ClassVar[str] = "list_directory"
     description: ClassVar[str] = "List contents of a directory in the codebase"
     args_schema: ClassVar[type[BaseModel]] = ListDirectoryInput
     codebase: Codebase = Field(exclude=True)
 
     def __init__(self, codebase: Codebase) -> None:
@@ -83,9 +89,9 @@
 class SearchTool(BaseTool):
     """Tool for searching the codebase."""
 
     name: ClassVar[str] = "search"
     description: ClassVar[str] = "Search the codebase using text search"
     args_schema: ClassVar[type[BaseModel]] = SearchInput
     codebase: Codebase = Field(exclude=True)
 
     def __init__(self, codebase: Codebase) -> None:
@@ -106,7 +112,7 @@
 class EditFileTool(BaseTool):
     """Tool for editing files."""
 
     name: ClassVar[str] = "edit_file"
     description: ClassVar[str] = "Edit a file by replacing its entire content"
     args_schema: ClassVar[type[BaseModel]] = EditFileInput
     codebase: Codebase = Field(exclude=True)
@@ -184,7 +190,10 @@
 
     symbol_name: str = Field(..., description="Name of the symbol to analyze")
     degree: int = Field(default=1, description="How many degrees of separation to traverse")
-    max_tokens: Optional[int] = Field(default=None, description="Optional maximum number of tokens for all source code combined")
+    max_tokens: Optional[int] = Field(
+        default=None,
+        description="Optional maximum number of tokens for all source code combined",
+    )
     collect_dependencies: bool = Field(default=True, description="Whether to collect dependencies")
     collect_usages: bool = Field(default=True, description="Whether to collect usages")
 
@@ -281,7 +290,10 @@
     source_file: str = Field(..., description="Path to the file containing the symbol")
     symbol_name: str = Field(..., description="Name of the symbol to move")
     target_file: str = Field(..., description="Path to the destination file")
-    strategy: Literal["update_all_imports", "add_back_edge"] = Field(default="update_all_imports", description="Strategy for handling imports: 'update_all_imports' (default) or 'add_back_edge'")
+    strategy: Literal["update_all_imports", "add_back_edge"] = Field(
+        default="update_all_imports",
+        description="Strategy for handling imports: 'update_all_imports' (default) or 'add_back_edge'",
+    )
     include_dependencies: bool = Field(default=True, description="Whether to move dependencies along with the symbol")
 
 
@@ -453,6 +465,73 @@
         return json.dumps(result, indent=2)
 
 
+class LinearGetIssueInput(BaseModel):
+    """Input for getting a Linear issue."""
+
+    issue_id: str = Field(..., description="ID of the Linear issue to retrieve")
+
+
+class LinearGetIssueTool(BaseTool):
+    """Tool for getting Linear issue details."""
+
+    name: ClassVar[str] = "linear_get_issue"
+    description: ClassVar[str] = "Get details of a Linear issue by its ID"
+    args_schema: ClassVar[type[BaseModel]] = LinearGetIssueInput
+    client: LinearClient = Field(exclude=True)
+
+    def __init__(self, client: LinearClient) -> None:
+        super().__init__(client=client)
+
+    def _run(self, issue_id: str) -> str:
+        result = linear_get_issue_tool(self.client, issue_id)
+        return json.dumps(result, indent=2)
+
+
+class LinearGetIssueCommentsInput(BaseModel):
+    """Input for getting Linear issue comments."""
+
+    issue_id: str = Field(..., description="ID of the Linear issue to get comments for")
+
+
+class LinearGetIssueCommentsTool(BaseTool):
+    """Tool for getting Linear issue comments."""
+
+    name: ClassVar[str] = "linear_get_issue_comments"
+    description: ClassVar[str] = "Get all comments on a Linear issue"
+    args_schema: ClassVar[type[BaseModel]] = LinearGetIssueCommentsInput
+    client: LinearClient = Field(exclude=True)
+
+    def __init__(self, client: LinearClient) -> None:
+        super().__init__(client=client)
+
+    def _run(self, issue_id: str) -> str:
+        result = linear_get_issue_comments_tool(self.client, issue_id)
+        return json.dumps(result, indent=2)
+
+
+class LinearCommentOnIssueInput(BaseModel):
+    """Input for commenting on a Linear issue."""
+
+    issue_id: str = Field(..., description="ID of the Linear issue to comment on")
+    body: str = Field(..., description="The comment text")
+
+
+class LinearCommentOnIssueTool(BaseTool):
+    """Tool for commenting on Linear issues."""
+
+    name: ClassVar[str] = "linear_comment_on_issue"
+    description: ClassVar[str] = "Add a comment to a Linear issue"
+    args_schema: ClassVar[type[BaseModel]] = LinearCommentOnIssueInput
+    client: LinearClient = Field(exclude=True)
+
+    def __init__(self, client: LinearClient) -> None:
+        super().__init__(client=client)
+
+    def _run(self, issue_id: str, body: str) -> str:
+        result = linear_comment_on_issue_tool(self.client, issue_id, body)
+        return json.dumps(result, indent=2)
+
+
 def get_workspace_tools(codebase: Codebase) -> list["BaseTool"]:
     """Get all workspace tools initialized with a codebase.
 
@@ -479,4 +558,7 @@
         SemanticEditTool(codebase),
         SemanticSearchTool(codebase),
         ViewFileTool(codebase),
+        LinearGetIssueTool(codebase),
+        LinearGetIssueCommentsTool(codebase),
+        LinearCommentOnIssueTool(codebase),
     ]

src/codegen/extensions/linear/__init__.py

@@ -0,0 +1,3 @@
+from .linear_client import LinearClient
+
+__all__ = ["LinearClient"]

src/codegen/extensions/linear/linear_client.py

@@ -0,0 +1,153 @@
+import json
+import logging
+import os
+from typing import Optional
+
+import requests
+from pydantic import BaseModel
+
+logger = logging.getLogger(__name__)
+
+
+# --- TYPES
+
+
+class LinearUser(BaseModel):
+    id: str
+    name: str
+
+
+class LinearComment(BaseModel):
+    id: str
+    body: str
+    user: LinearUser | None = None
+
+
+class LinearIssue(BaseModel):
+    id: str
+    title: str
+    description: str | None = None
+
+
+class LinearClient:
+    api_headers: dict
+    api_endpoint = "https://api.linear.app/graphql"
+
+    def __init__(self, access_token: Optional[str] = None):
+        if not access_token:
+            access_token = os.getenv("LINEAR_ACCESS_TOKEN")
+            if not access_token:
+                msg = "access_token is required"
+                raise ValueError(msg)
+        self.access_token = access_token
+        self.api_headers = {
+            "Content-Type": "application/json",
+            "Authorization": self.access_token,
+        }
+
+    def get_issue(self, issue_id: str) -> LinearIssue:
+        query = """
+            query getIssue($issueId: String!) {
+                issue(id: $issueId) {
+                    id
+                    title
+                    description
+                }
+            }
+        """
+        variables = {"issueId": issue_id}
+        response = requests.post(self.api_endpoint, headers=self.api_headers, json={"query": query, "variables": variables})
+        data = response.json()
+        issue_data = data["data"]["issue"]
+        return LinearIssue(id=issue_data["id"], title=issue_data["title"], description=issue_data["description"])
+
+    def get_issue_comments(self, issue_id: str) -> list[LinearComment]:
+        query = """
+            query getIssueComments($issueId: String!) {
+                issue(id: $issueId) {
+                    comments {
+                    nodes {
+                        id
+                        body
+                        user {
+                            id
+                            name
+                        }
+                    }
+
+                    }
+                }
+            }
+        """
+        variables = {"issueId": issue_id}
+        response = requests.post(self.api_endpoint, headers=self.api_headers, json={"query": query, "variables": variables})
+        data = response.json()
+        comments = data["data"]["issue"]["comments"]["nodes"]
+
+        # Parse comments into list of LinearComment objects
+        parsed_comments = []
+        for comment in comments:
+            user = comment.get("user", None)
+            parsed_comment = LinearComment(id=comment["id"], body=comment["body"], user=LinearUser(id=user.get("id"), name=user.get("name")) if user else None)
+            parsed_comments.append(parsed_comment)
+
+        # Convert raw comments to LinearComment objects
+        return parsed_comments
+
+    def comment_on_issue(self, issue_id: str, body: str) -> dict:
+        """issue_id is our internal issue ID"""
+        query = """mutation makeComment($issueId: String!, $body: String!) {
+          commentCreate(input: {issueId: $issueId, body: $body}) {
+            comment {
+              id
+              body
+              url
+              user {
+                id
+                name
+              }
+            }
+          }
+        }
+        """
+        variables = {"issueId": issue_id, "body": body}
+        response = requests.post(
+            self.api_endpoint,
+            headers=self.api_headers,
+            data=json.dumps({"query": query, "variables": variables}),
+        )
+        data = response.json()
+        try:
+            comment_data = data["data"]["commentCreate"]["comment"]
+
+            return comment_data
+        except Exception as e:
+            msg = f"Error creating comment\n{data}\n{e}"
+            raise Exception(msg)
+
+    def register_webhook(self, webhook_url: str, team_id: str, secret: str, enabled: bool, resource_types: list[str]):
+        mutation = """
+            mutation createWebhook($input: WebhookCreateInput!) {
+                webhookCreate(input: $input) {
+                    success
+                    webhook {
+                    id
+                    enabled
+                    }
+                }
+            }
+        """
+
+        variables = {
+            "input": {
+                "url": webhook_url,
+                "teamId": team_id,
+                "resourceTypes": resource_types,
+                "enabled": enabled,
+                "secret": secret,
+            }
+        }
+
+        response = requests.post(self.api_endpoint, headers=self.api_headers, json={"query": mutation, "variables": variables})
+        body = response.json()
+        return body

src/codegen/extensions/tools/__init__.py

@@ -8,6 +8,12 @@
 from .github.create_pr_comment import create_pr_comment
 from .github.create_pr_review_comment import create_pr_review_comment
 from .github.view_pr import view_pr
+from .linear_tools import (
+    linear_comment_on_issue_tool,
+    linear_get_issue_comments_tool,
+    linear_get_issue_tool,
+    linear_register_webhook_tool,
+)
 from .list_directory import list_directory
 from .move_symbol import move_symbol
 from .rename_file import rename_file
@@ -27,6 +33,11 @@
     "create_pr_review_comment",
     "delete_file",
     "edit_file",
+    # Linear operations
+    "linear_comment_on_issue_tool",
+    "linear_get_issue_comments_tool",
+    "linear_get_issue_tool",
+    "linear_register_webhook_tool",
     "list_directory",
     # Symbol operations
     "move_symbol",

src/codegen/extensions/tools/linear_tools.py

@@ -0,0 +1,39 @@
+from typing import Any
+
+from codegen.extensions.linear.linear_client import LinearClient
+
+
+def linear_get_issue_tool(client: LinearClient, issue_id: str) -> dict[str, Any]:
+    """Get an issue by its ID."""
+    try:
+        issue = client.get_issue(issue_id)
+        return {"status": "success", "issue": issue.dict()}
+    except Exception as e:
+        return {"error": f"Failed to get issue: {e!s}"}
+
+
+def linear_get_issue_comments_tool(client: LinearClient, issue_id: str) -> dict[str, Any]:
+    """Get comments for a specific issue."""
+    try:
+        comments = client.get_issue_comments(issue_id)
+        return {"status": "success", "comments": [comment.dict() for comment in comments]}
+    except Exception as e:
+        return {"error": f"Failed to get issue comments: {e!s}"}
+
+
+def linear_comment_on_issue_tool(client: LinearClient, issue_id: str, body: str) -> dict[str, Any]:
+    """Add a comment to an issue."""
+    try:
+        comment = client.comment_on_issue(issue_id, body)
+        return {"status": "success", "comment": comment}
+    except Exception as e:
+        return {"error": f"Failed to comment on issue: {e!s}"}
+
+
+def linear_register_webhook_tool(client: LinearClient, webhook_url: str, team_id: str, secret: str, enabled: bool, resource_types: list[str]) -> dict[str, Any]:
+    """Register a webhook with Linear."""
+    try:
+        response = client.register_webhook(webhook_url, team_id, secret, enabled, resource_types)
+        return {"status": "success", "response": response}
+    except Exception as e:
+        return {"error": f"Failed to register webhook: {e!s}"}

tests/integration/extension/test_linear.py

@@ -0,0 +1,43 @@
+"""Tests for Linear tools."""
+
+import os
+
+import pytest
+
+from codegen.extensions.linear.linear_client import LinearClient
+from codegen.extensions.tools.linear_tools import (
+    linear_comment_on_issue_tool,
+    linear_get_issue_comments_tool,
+    linear_get_issue_tool,
+)
+
+
+@pytest.fixture
+def client() -> LinearClient:
+    """Create a Linear client for testing."""
+    token = os.getenv("LINEAR_ACCESS_TOKEN")
+    if not token:
+        pytest.skip("LINEAR_ACCESS_TOKEN environment variable not set")
+    return LinearClient(token)
+
+
+def test_linear_get_issue(client: LinearClient) -> None:
+    """Test getting an issue from Linear."""
+    # Link to issue: https://linear.app/codegen-sh/issue/CG-10775/read-file-and-reveal-symbol-tool-size-limits
+    issue = linear_get_issue_tool(client, "CG-10775")
+    assert issue["status"] == "success"
+    assert issue["issue"]["id"] == "d5a7d6db-e20d-4d67-98f8-acedef6d3536"
+
+
+def test_linear_get_issue_comments(client: LinearClient) -> None:
+    """Test getting comments for an issue from Linear."""
+    comments = linear_get_issue_comments_tool(client, "CG-10775")
+    assert comments["status"] == "success"
+    assert len(comments["comments"]) > 1
+
+
+def test_linear_comment_on_issue(client: LinearClient) -> None:
+    """Test commenting on a Linear issue."""
+    test_comment = "Test comment from automated testing"
+    result = linear_comment_on_issue_tool(client, "CG-10775", test_comment)
+    assert result["status"] == "success"