feat: add LinearTool for Linear project management API

2026-05-07 10:12:38 +00:00 · 2026-04-20 15:08:04 -03:00
parent 01b8437940
commit 7961678879
1 changed files with 124 additions and 0 deletions
--- a/crewai/tools/linear_tool.py
+++ b/crewai/tools/linear_tool.py
@@ -0,0 +1,124 @@
+import os
+from enum import Enum
+from typing import Any, Type
+
+import httpx
+from crewai.tools import BaseTool
+from pydantic import BaseModel, Field
+
+LINEAR_API_URL = "https://api.linear.app/graphql"
+
+
+class LinearAction(str, Enum):
+    MY_ISSUES = "my_issues"
+    LIST_TEAMS = "list_teams"
+    LIST_PROJECTS = "list_projects"
+
+
+class LinearToolInput(BaseModel):
+    action: LinearAction = Field(
+        description=(
+            "Action to perform: "
+            "'my_issues' — fetch issues assigned to the authenticated user; "
+            "'list_teams' — list all teams in the workspace; "
+            "'list_projects' — list all projects in the workspace."
+        )
+    )
+    first: int = Field(
+        default=25,
+        ge=1,
+        le=250,
+        description="Maximum number of records to return (1–250).",
+    )
+
+
+_QUERIES: dict[LinearAction, str] = {
+    LinearAction.MY_ISSUES: """
+        query MyIssues($first: Int!) {
+          viewer {
+            assignedIssues(first: $first, orderBy: updatedAt) {
+              nodes {
+                id
+                identifier
+                title
+                state { name }
+                priority
+                url
+                updatedAt
+              }
+            }
+          }
+        }
+    """,
+    LinearAction.LIST_TEAMS: """
+        query ListTeams($first: Int!) {
+          teams(first: $first) {
+            nodes {
+              id
+              name
+              key
+              description
+            }
+          }
+        }
+    """,
+    LinearAction.LIST_PROJECTS: """
+        query ListProjects($first: Int!) {
+          projects(first: $first, orderBy: updatedAt) {
+            nodes {
+              id
+              name
+              description
+              state
+              url
+              updatedAt
+            }
+          }
+        }
+    """,
+}
+
+
+def _extract(action: LinearAction, data: dict) -> list[dict]:
+    if action == LinearAction.MY_ISSUES:
+        return data["viewer"]["assignedIssues"]["nodes"]
+    if action == LinearAction.LIST_TEAMS:
+        return data["teams"]["nodes"]
+    if action == LinearAction.LIST_PROJECTS:
+        return data["projects"]["nodes"]
+    return []
+
+
+class LinearTool(BaseTool):
+    name: str = "Linear API Tool"
+    description: str = (
+        "Interact with the Linear project management API. "
+        "Supports fetching your assigned issues, listing teams, and listing projects."
+    )
+    args_schema: Type[BaseModel] = LinearToolInput
+
+    def _run(self, action: LinearAction, first: int = 25) -> Any:
+        api_key = os.environ.get("LINEAR_API_KEY", "")
+        if not api_key:
+            raise EnvironmentError("LINEAR_API_KEY environment variable is not set.")
+
+        query = _QUERIES[action]
+        payload = {"query": query, "variables": {"first": first}}
+        headers = {
+            "Authorization": api_key,
+            "Content-Type": "application/json",
+        }
+
+        response = httpx.post(
+            LINEAR_API_URL,
+            json=payload,
+            headers=headers,
+            timeout=15,
+        )
+        response.raise_for_status()
+
+        body = response.json()
+        if "errors" in body:
+            raise RuntimeError(f"Linear API errors: {body['errors']}")
+
+        return _extract(action, body["data"])