Compare commits

...

2 Commits

Author SHA1 Message Date
iris-clawd
75ef1ecf91 test: add test script for LinearTool 2026-04-20 15:08:13 -03:00
iris-clawd
7961678879 feat: add LinearTool for Linear project management API 2026-04-20 15:08:04 -03:00
2 changed files with 181 additions and 0 deletions

124
crewai/tools/linear_tool.py Normal file
View File

@@ -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 (1250).",
)
_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"])

View File

@@ -0,0 +1,57 @@
"""
Test script for LinearTool — runs against the real Linear API.
Usage:
LINEAR_API_KEY=lin_api_xxxxxxxxxxxx python tests/tools/test_linear_tool.py
Set LINEAR_API_KEY to your actual Personal API key from:
https://linear.app/settings/api (Profile → API → Personal API keys)
"""
import json
import os
import sys
from crewai.tools.linear_tool import LinearAction, LinearTool
def pretty(data: object) -> str:
return json.dumps(data, indent=2, default=str)
def run_test(tool: LinearTool, label: str, action: LinearAction, first: int = 5) -> None:
print(f"\n{'' * 60}")
print(f" {label}")
print(f"{'' * 60}")
try:
result = tool._run(action=action, first=first)
if not result:
print(" (no records returned)")
else:
print(pretty(result))
except Exception as exc:
print(f" ERROR: {exc}", file=sys.stderr)
def main() -> None:
if not os.environ.get("LINEAR_API_KEY"):
print(
"ERROR: Set LINEAR_API_KEY before running.\n"
" export LINEAR_API_KEY=lin_api_xxxxxxxxxxxx",
file=sys.stderr,
)
sys.exit(1)
tool = LinearTool()
run_test(tool, "My assigned issues (up to 5)", LinearAction.MY_ISSUES, first=5)
run_test(tool, "Teams (up to 10)", LinearAction.LIST_TEAMS, first=10)
run_test(tool, "Projects (up to 10)", LinearAction.LIST_PROJECTS, first=10)
print(f"\n{'' * 60}")
print(" All tests complete.")
print(f"{'' * 60}\n")
if __name__ == "__main__":
main()