feat: add linear tools for listing teams, get/create/update/delete issues, and get issue comments (#1105)

Adds 6 new agent tools backed by Linear's GraphQL API, with a shared
_graphql_request helper to reduce boilerplate. Refactors existing
comment_on_linear_issue to use the same helper.

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

agent/server.py

@@ -39,6 +39,12 @@ from .tools import (
     github_comment,
     http_request,
     linear_comment,
+    linear_create_issue,
+    linear_delete_issue,
+    linear_get_issue,
+    linear_get_issue_comments,
+    linear_list_teams,
+    linear_update_issue,
     slack_thread_reply,
 )
 from .utils.auth import resolve_github_token
@@ -399,6 +405,12 @@ async def get_agent(config: RunnableConfig) -> Pregel:  # noqa: PLR0915
             fetch_url,
             commit_and_open_pr,
             linear_comment,
+            linear_create_issue,
+            linear_delete_issue,
+            linear_get_issue,
+            linear_get_issue_comments,
+            linear_list_teams,
+            linear_update_issue,
             slack_thread_reply,
             github_comment,
         ],

agent/tools/__init__.py

@@ -3,6 +3,12 @@ from .fetch_url import fetch_url
 from .github_comment import github_comment
 from .http_request import http_request
 from .linear_comment import linear_comment
+from .linear_create_issue import linear_create_issue
+from .linear_delete_issue import linear_delete_issue
+from .linear_get_issue import linear_get_issue
+from .linear_get_issue_comments import linear_get_issue_comments
+from .linear_list_teams import linear_list_teams
+from .linear_update_issue import linear_update_issue
 from .slack_thread_reply import slack_thread_reply
 
 __all__ = [
@@ -11,5 +17,11 @@ __all__ = [
     "github_comment",
     "http_request",
     "linear_comment",
+    "linear_create_issue",
+    "linear_delete_issue",
+    "linear_get_issue",
+    "linear_get_issue_comments",
+    "linear_list_teams",
+    "linear_update_issue",
     "slack_thread_reply",
 ]

agent/tools/linear_create_issue.py

@@ -0,0 +1,43 @@
+import asyncio
+from typing import Any
+
+from ..utils.linear import create_issue
+
+
+def linear_create_issue(
+    team_id: str,
+    title: str,
+    description: str | None = None,
+    assignee_id: str | None = None,
+    priority: int | None = None,
+    state_id: str | None = None,
+    label_ids: list[str] | None = None,
+    project_id: str | None = None,
+) -> dict[str, Any]:
+    """Create a new Linear issue.
+
+    Args:
+        team_id: The ID of the team to create the issue in.
+        title: The title of the issue.
+        description: Optional markdown description.
+        assignee_id: Optional user ID to assign the issue to.
+        priority: Optional priority (0=none, 1=urgent, 2=high, 3=medium, 4=low).
+        state_id: Optional workflow state ID.
+        label_ids: Optional list of label IDs to apply.
+        project_id: Optional project ID to associate with.
+
+    Returns:
+        Dictionary with 'success' bool and 'issue' details.
+    """
+    return asyncio.run(
+        create_issue(
+            team_id=team_id,
+            title=title,
+            description=description,
+            assignee_id=assignee_id,
+            priority=priority,
+            state_id=state_id,
+            label_ids=label_ids,
+            project_id=project_id,
+        )
+    )

agent/tools/linear_delete_issue.py

@@ -0,0 +1,16 @@
+import asyncio
+from typing import Any
+
+from ..utils.linear import delete_issue
+
+
+def linear_delete_issue(issue_id: str) -> dict[str, Any]:
+    """Delete a Linear issue.
+
+    Args:
+        issue_id: The Linear issue UUID to delete.
+
+    Returns:
+        Dictionary with 'success' bool.
+    """
+    return asyncio.run(delete_issue(issue_id))

agent/tools/linear_get_issue.py

@@ -0,0 +1,16 @@
+import asyncio
+from typing import Any
+
+from ..utils.linear import get_issue
+
+
+def linear_get_issue(issue_id: str) -> dict[str, Any]:
+    """Get a Linear issue by its ID.
+
+    Args:
+        issue_id: The Linear issue UUID.
+
+    Returns:
+        Dictionary with 'issue' containing full issue details.
+    """
+    return asyncio.run(get_issue(issue_id))

agent/tools/linear_get_issue_comments.py

@@ -0,0 +1,16 @@
+import asyncio
+from typing import Any
+
+from ..utils.linear import get_issue_comments
+
+
+def linear_get_issue_comments(issue_id: str) -> dict[str, Any]:
+    """Get all comments on a Linear issue.
+
+    Args:
+        issue_id: The Linear issue UUID.
+
+    Returns:
+        Dictionary with 'comments' list, each containing id, body, createdAt, user, etc.
+    """
+    return asyncio.run(get_issue_comments(issue_id))

agent/tools/linear_list_teams.py

@@ -0,0 +1,13 @@
+import asyncio
+from typing import Any
+
+from ..utils.linear import list_teams
+
+
+def linear_list_teams() -> dict[str, Any]:
+    """List all teams in the Linear workspace.
+
+    Returns:
+        Dictionary with 'teams' list, each containing id, name, key, and description.
+    """
+    return asyncio.run(list_teams())

agent/tools/linear_update_issue.py

@@ -0,0 +1,40 @@
+import asyncio
+from typing import Any
+
+from ..utils.linear import update_issue
+
+
+def linear_update_issue(
+    issue_id: str,
+    title: str | None = None,
+    description: str | None = None,
+    assignee_id: str | None = None,
+    priority: int | None = None,
+    state_id: str | None = None,
+    label_ids: list[str] | None = None,
+) -> dict[str, Any]:
+    """Update an existing Linear issue.
+
+    Args:
+        issue_id: The Linear issue UUID to update.
+        title: New title for the issue.
+        description: New markdown description.
+        assignee_id: User ID to assign the issue to.
+        priority: Priority (0=none, 1=urgent, 2=high, 3=medium, 4=low).
+        state_id: Workflow state ID to transition to.
+        label_ids: List of label IDs to set.
+
+    Returns:
+        Dictionary with 'success' bool and updated 'issue' details.
+    """
+    return asyncio.run(
+        update_issue(
+            issue_id=issue_id,
+            title=title,
+            description=description,
+            assignee_id=assignee_id,
+            priority=priority,
+            state_id=state_id,
+            label_ids=label_ids,
+        )
+    )

agent/utils/linear.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import logging
 import os
+from typing import Any
 
 import httpx
 
@@ -12,59 +13,54 @@ from agent.utils.langsmith import get_langsmith_trace_url
 logger = logging.getLogger(__name__)
 
 LINEAR_API_KEY = os.environ.get("LINEAR_API_KEY", "")
+LINEAR_API_URL = "https://api.linear.app/graphql"
+
+
+def _headers() -> dict[str, str]:
+    return {
+        "Authorization": LINEAR_API_KEY,
+        "Content-Type": "application/json",
+    }
+
+
+async def _graphql_request(query: str, variables: dict[str, Any] | None = None) -> dict[str, Any]:
+    """Execute a GraphQL request against the Linear API."""
+    if not LINEAR_API_KEY:
+        return {"error": "LINEAR_API_KEY is not set"}
+
+    async with httpx.AsyncClient() as http_client:
+        try:
+            response = await http_client.post(
+                LINEAR_API_URL,
+                headers=_headers(),
+                json={"query": query, "variables": variables or {}},
+            )
+            response.raise_for_status()
+            result = response.json()
+            if result.get("errors"):
+                return {"error": result["errors"]}
+            return result.get("data", {})
+        except Exception as e:  # noqa: BLE001
+            return {"error": str(e)}
 
 
 async def comment_on_linear_issue(
     issue_id: str, comment_body: str, parent_id: str | None = None
 ) -> bool:
-    """Add a comment to a Linear issue, optionally as a reply to a specific comment.
+    """Add a comment to a Linear issue, optionally as a reply to a specific comment."""
-
-    Args:
-        issue_id: The Linear issue ID
-        comment_body: The comment text
-        parent_id: Optional comment ID to reply to
-
-    Returns:
-        True if successful, False otherwise
-    """
-    if not LINEAR_API_KEY:
-        return False
-
-    url = "https://api.linear.app/graphql"
-
     mutation = """
     mutation CommentCreate($issueId: String!, $body: String!, $parentId: String) {
         commentCreate(input: { issueId: $issueId, body: $body, parentId: $parentId }) {
             success
-            comment {
+            comment { id }
-                id
-            }
         }
     }
     """
-
+    result = await _graphql_request(
-    async with httpx.AsyncClient() as http_client:
+        mutation,
-        try:
+        {"issueId": issue_id, "body": comment_body, "parentId": parent_id},
-            response = await http_client.post(
+    )
-                url,
+    return bool(result.get("commentCreate", {}).get("success"))
-                headers={
-                    "Authorization": LINEAR_API_KEY,
-                    "Content-Type": "application/json",
-                },
-                json={
-                    "query": mutation,
-                    "variables": {
-                        "issueId": issue_id,
-                        "body": comment_body,
-                        "parentId": parent_id,
-                    },
-                },
-            )
-            response.raise_for_status()
-            result = response.json()
-            return bool(result.get("data", {}).get("commentCreate", {}).get("success"))
-        except Exception:  # noqa: BLE001
-            return False
 
 
 async def post_linear_trace_comment(issue_id: str, run_id: str, triggering_comment_id: str) -> None:
@@ -76,3 +72,190 @@ async def post_linear_trace_comment(issue_id: str, run_id: str, triggering_comme
             f"On it! [View trace]({trace_url})",
             parent_id=triggering_comment_id or None,
         )
+
+
+async def list_teams() -> dict[str, Any]:
+    """List all teams in the Linear workspace."""
+    query = """
+    query {
+        teams {
+            nodes {
+                id
+                name
+                key
+                description
+            }
+        }
+    }
+    """
+    result = await _graphql_request(query)
+    if "error" in result:
+        return result
+    return {"teams": result.get("teams", {}).get("nodes", [])}
+
+
+async def get_issue(issue_id: str) -> dict[str, Any]:
+    """Get a Linear issue by ID."""
+    query = """
+    query GetIssue($id: String!) {
+        issue(id: $id) {
+            id
+            identifier
+            title
+            description
+            priority
+            priorityLabel
+            state { id name }
+            assignee { id name email }
+            team { id name key }
+            project { id name }
+            labels { nodes { id name } }
+            createdAt
+            updatedAt
+            url
+        }
+    }
+    """
+    result = await _graphql_request(query, {"id": issue_id})
+    if "error" in result:
+        return result
+    return {"issue": result.get("issue")}
+
+
+async def create_issue(
+    team_id: str,
+    title: str,
+    description: str | None = None,
+    assignee_id: str | None = None,
+    priority: int | None = None,
+    state_id: str | None = None,
+    label_ids: list[str] | None = None,
+    project_id: str | None = None,
+) -> dict[str, Any]:
+    """Create a new Linear issue."""
+    mutation = """
+    mutation IssueCreate($input: IssueCreateInput!) {
+        issueCreate(input: $input) {
+            success
+            issue {
+                id
+                identifier
+                title
+                url
+            }
+        }
+    }
+    """
+    input_vars: dict[str, Any] = {"teamId": team_id, "title": title}
+    if description is not None:
+        input_vars["description"] = description
+    if assignee_id is not None:
+        input_vars["assigneeId"] = assignee_id
+    if priority is not None:
+        input_vars["priority"] = priority
+    if state_id is not None:
+        input_vars["stateId"] = state_id
+    if label_ids is not None:
+        input_vars["labelIds"] = label_ids
+    if project_id is not None:
+        input_vars["projectId"] = project_id
+
+    result = await _graphql_request(mutation, {"input": input_vars})
+    if "error" in result:
+        return result
+    issue_create = result.get("issueCreate", {})
+    return {
+        "success": issue_create.get("success", False),
+        "issue": issue_create.get("issue"),
+    }
+
+
+async def get_issue_comments(issue_id: str) -> dict[str, Any]:
+    """Get comments for a Linear issue."""
+    query = """
+    query GetIssueComments($id: String!) {
+        issue(id: $id) {
+            comments {
+                nodes {
+                    id
+                    body
+                    createdAt
+                    updatedAt
+                    user { id name email }
+                }
+            }
+        }
+    }
+    """
+    result = await _graphql_request(query, {"id": issue_id})
+    if "error" in result:
+        return result
+    issue = result.get("issue")
+    if not issue:
+        return {"error": f"Issue {issue_id} not found"}
+    return {"comments": issue.get("comments", {}).get("nodes", [])}
+
+
+async def update_issue(
+    issue_id: str,
+    title: str | None = None,
+    description: str | None = None,
+    assignee_id: str | None = None,
+    priority: int | None = None,
+    state_id: str | None = None,
+    label_ids: list[str] | None = None,
+) -> dict[str, Any]:
+    """Update an existing Linear issue."""
+    mutation = """
+    mutation IssueUpdate($id: String!, $input: IssueUpdateInput!) {
+        issueUpdate(id: $id, input: $input) {
+            success
+            issue {
+                id
+                identifier
+                title
+                url
+            }
+        }
+    }
+    """
+    input_vars: dict[str, Any] = {}
+    if title is not None:
+        input_vars["title"] = title
+    if description is not None:
+        input_vars["description"] = description
+    if assignee_id is not None:
+        input_vars["assigneeId"] = assignee_id
+    if priority is not None:
+        input_vars["priority"] = priority
+    if state_id is not None:
+        input_vars["stateId"] = state_id
+    if label_ids is not None:
+        input_vars["labelIds"] = label_ids
+
+    if not input_vars:
+        return {"error": "No fields to update"}
+
+    result = await _graphql_request(mutation, {"id": issue_id, "input": input_vars})
+    if "error" in result:
+        return result
+    issue_update = result.get("issueUpdate", {})
+    return {
+        "success": issue_update.get("success", False),
+        "issue": issue_update.get("issue"),
+    }
+
+
+async def delete_issue(issue_id: str) -> dict[str, Any]:
+    """Delete a Linear issue."""
+    mutation = """
+    mutation IssueDelete($id: String!) {
+        issueDelete(id: $id) {
+            success
+        }
+    }
+    """
+    result = await _graphql_request(mutation, {"id": issue_id})
+    if "error" in result:
+        return result
+    return {"success": result.get("issueDelete", {}).get("success", False)}