admin-valentin/arcade-mcp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add Simplified Linear Toolkit (#465)

Browse source 
This PR adds a simplified Linear toolkit focused on core functionality:

## What's Included

**Tools:**
- `get_issue` - Get detailed information about Linear issues
- `get_teams` - Get team information and details

**Models:**
- `DateRange` enum with comprehensive date range support (TODAY,
YESTERDAY, THIS_WEEK, LAST_WEEK, THIS_MONTH, LAST_MONTH, THIS_YEAR,
LAST_YEAR, LAST_7_DAYS, LAST_30_DAYS)
- Timezone-aware datetime handling following Google toolkit patterns

## What's Simplified

This toolkit has been streamlined by removing:
- Cycles management tools
- Projects management tools  
- Users management tools
- Workflows management tools
- Corresponding tests and evaluations for removed features

## Quality Assurance

- All linting and formatting checks pass
- Comprehensive test coverage for included functionality
- Follows established patterns from Google toolkit

---------

Co-authored-by: Eric Gustin <34000337+EricGustin@users.noreply.github.com>
This commit is contained in:
Shub 2025-07-18 14:08:58 -07:00 committed by GitHub
parent cb7b386a43
commit 0b1f513826
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 2524 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
docker/docker.engine.yaml
View file

@ -133,6 +133,14 @@ auth:
client_id: ${env:LINKEDIN_CLIENT_ID}
client_secret: ${env:LINKEDIN_CLIENT_SECRET}
- id: arcade-linear
description: "The Linear provider for Arcade"
enabled: false
type: oauth2
provider_id: linear
client_id: ${env:LINEAR_CLIENT_ID}
client_secret: ${env:LINEAR_CLIENT_SECRET}
- id: default-microsoft
description: "The default Microsoft provider"
enabled: false

3
docker/env.example
View file

@ -15,6 +15,9 @@ GOOGLE_CLIENT_SECRET=
LINKEDIN_CLIENT_ID=
LINKEDIN_CLIENT_SECRET=""
LINEAR_CLIENT_ID=
LINEAR_CLIENT_SECRET=
MICROSOFT_CLIENT_ID=
MICROSOFT_CLIENT_SECRET=""

1
docker/toolkits.txt
View file

@ -6,6 +6,7 @@ arcade-github
arcade-google
arcade-hubspot
arcade-jira
arcade-linear
arcade-linkedin
arcade-math
arcade-microsoft

18
toolkits/linear/.pre-commit-config.yaml Normal file
View file

@ -0,0 +1,18 @@
files: ^.*/linear/.*
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: "v4.4.0"
hooks:
- id: check-case-conflict
- id: check-merge-conflict
- id: check-toml
- id: check-yaml
- id: end-of-file-fixer
- id: trailing-whitespace
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.6.7
hooks:
- id: ruff
args: [--fix]
- id: ruff-format

45
toolkits/linear/.ruff.toml Normal file
View file

@ -0,0 +1,45 @@
target-version = "py310"
line-length = 100
fix = true
[lint]
select = [
# flake8-2020
"YTT",
# flake8-bandit
"S",
# flake8-bugbear
"B",
# flake8-builtins
"A",
# flake8-comprehensions
"C4",
# flake8-debugger
"T10",
# flake8-simplify
"SIM",
# isort
"I",
# mccabe
"C90",
# pycodestyle
"E", "W",
# pyflakes
"F",
# pygrep-hooks
"PGH",
# pyupgrade
"UP",
# ruff
"RUF",
# tryceratops
"TRY",
]
[lint.per-file-ignores]
"*" = ["TRY003"]
"**/tests/*" = ["S101"]
[format]
preview = true
skip-magic-trailing-comma = false

21
toolkits/linear/LICENSE Normal file
View file

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Arcade
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

55
toolkits/linear/Makefile Normal file
View file

@ -0,0 +1,55 @@
.PHONY: help
help:
@echo "🛠️ Linear Commands:\n"
@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'
.PHONY: install
install: ## Install the uv environment and install all packages with dependencies
@echo "🚀 Creating virtual environment and installing all packages using uv"
@uv sync --active --all-extras --no-sources
@if [ -f .pre-commit-config.yaml ]; then uv run --no-sources pre-commit install; fi
@echo "✅ All packages and dependencies installed via uv"
.PHONY: install-local
install-local: ## Install the uv environment and install all packages with dependencies with local Arcade sources
@echo "🚀 Creating virtual environment and installing all packages using uv"
@uv sync --active --all-extras
@if [ -f .pre-commit-config.yaml ]; then uv run pre-commit install; fi
@echo "✅ All packages and dependencies installed via uv"
.PHONY: build
build: clean-build ## Build wheel file using poetry
@echo "🚀 Creating wheel file"
uv build
.PHONY: clean-build
clean-build: ## clean build artifacts
@echo "🗑️ Cleaning dist directory"
rm -rf dist
.PHONY: test
test: ## Test the code with pytest
@echo "🚀 Testing code: Running pytest"
@uv run --no-sources pytest -W ignore -v --cov --cov-config=pyproject.toml --cov-report=xml
.PHONY: coverage
coverage: ## Generate coverage report
@echo "coverage report"
@uv run --no-sources coverage report
@echo "Generating coverage report"
@uv run --no-sources coverage html
.PHONY: bump-version
bump-version: ## Bump the version in the pyproject.toml file by a patch version
@echo "🚀 Bumping version in pyproject.toml"
uv version --no-sources --bump patch
.PHONY: check
check: ## Run code quality tools.
@if [ -f .pre-commit-config.yaml ]; then\
echo "🚀 Linting code: Running pre-commit";\
uv run --no-sources pre-commit run -a;\
fi
@echo "🚀 Static type checking: Running mypy"
@uv run --no-sources mypy --config-file=pyproject.toml

13
toolkits/linear/arcade_linear/__init__.py Normal file
View file

@ -0,0 +1,13 @@
"""Linear Toolkit for Arcade AI"""
from arcade_linear.models import DateRange
from arcade_linear.tools import (
get_issue, # Get specific issue details
get_teams, # Get team information
)
__all__ = [
"DateRange",
"get_issue",
"get_teams",
]

316
toolkits/linear/arcade_linear/client.py Normal file
View file

@ -0,0 +1,316 @@
import asyncio
import json
from dataclasses import dataclass
from typing import Any, cast
import httpx
from arcade_tdk.errors import ToolExecutionError
from arcade_linear.constants import (
LINEAR_API_URL,
LINEAR_MAX_CONCURRENT_REQUESTS,
LINEAR_MAX_TIMEOUT_SECONDS,
)
@dataclass
class LinearClient:
"""Client for interacting with Linear's GraphQL API"""
auth_token: str
api_url: str = LINEAR_API_URL
max_concurrent_requests: int = LINEAR_MAX_CONCURRENT_REQUESTS
timeout_seconds: int = LINEAR_MAX_TIMEOUT_SECONDS
_semaphore: asyncio.Semaphore | None = None
def __post_init__(self) -> None:
self._semaphore = self._semaphore or asyncio.Semaphore(self.max_concurrent_requests)
def _build_headers(self, additional_headers: dict[str, str] | None = None) -> dict[str, str]:
"""Build headers for GraphQL requests"""
headers = {
"Authorization": f"Bearer {self.auth_token}",
"Content-Type": "application/json",
"Accept": "application/json",
}
if additional_headers:
headers.update(additional_headers)
return headers
def _build_error_message(self, response: httpx.Response) -> tuple[str, str]:
"""Build user-friendly and developer error messages from response"""
try:
data = response.json()
if data.get("errors"):
errors = data["errors"]
if len(errors) == 1:
error = errors[0]
user_message = error.get("message", "Unknown GraphQL error")
dev_message = (
f"GraphQL error: {json.dumps(error)} (HTTP {response.status_code})"
)
else:
error_messages = [err.get("message", "Unknown error") for err in errors]
user_message = f"Multiple errors: {'; '.join(error_messages)}"
dev_message = (
f"Multiple GraphQL errors: {json.dumps(errors)} "
f"(HTTP {response.status_code})"
)
else:
user_message = f"HTTP {response.status_code}: {response.reason_phrase}"
dev_message = f"HTTP {response.status_code}: {response.text}"
except Exception as e:
user_message = "Failed to parse Linear API error response"
dev_message = (
f"Failed to parse error response: {type(e).__name__}: {e!s} | "
f"Raw response: {response.text}"
)
return user_message, dev_message
def _raise_for_status(self, response: httpx.Response) -> None:
"""Raise appropriate errors for non-200 responses"""
if response.status_code < 300:
# Check for GraphQL errors in successful HTTP responses
try:
data = response.json()
if data.get("errors"):
user_message, dev_message = self._build_error_message(response)
raise ToolExecutionError(user_message, developer_message=dev_message)
except (ValueError, KeyError):
# Response isn't JSON or doesn't have expected structure
pass
return
user_message, dev_message = self._build_error_message(response)
raise ToolExecutionError(user_message, developer_message=dev_message)
async def execute_query(
self, query: str, variables: dict[str, Any] | None = None, operation_name: str | None = None
) -> dict[str, Any]:
"""Execute a GraphQL query"""
payload: dict[str, Any] = {
"query": query.strip(),
}
if variables:
payload["variables"] = variables
if operation_name:
payload["operationName"] = operation_name
headers = self._build_headers()
async with self._semaphore, httpx.AsyncClient(timeout=self.timeout_seconds) as client: # type: ignore[union-attr]
response = await client.post(
self.api_url,
json=payload,
headers=headers,
)
self._raise_for_status(response)
return cast(dict[str, Any], response.json())
async def get_teams(
self,
first: int = 50,
after: str | None = None,
include_archived: bool = False,
name_filter: str | None = None,
) -> dict[str, Any]:
"""Get teams with optional filtering"""
query = """
query GetTeams($first: Int!, $after: String, $filter: TeamFilter) {
teams(first: $first, after: $after, filter: $filter) {
nodes {
id
key
name
description
private
archivedAt
createdAt
updatedAt
icon
color
cyclesEnabled
issueEstimationType
organization {
id
name
}
members {
nodes {
id
name
email
displayName
avatarUrl
}
}
}
pageInfo {
hasNextPage
hasPreviousPage
startCursor
endCursor
}
}
}
"""
# Build filter
team_filter = {}
if name_filter:
team_filter["name"] = {"containsIgnoreCase": name_filter}
variables = {"first": first, "after": after, "filter": team_filter if team_filter else None}
result = await self.execute_query(query, variables)
return cast(dict[str, Any], result["data"]["teams"])
async def get_issue_by_id(self, issue_id: str) -> dict[str, Any]:
"""Get a single issue by ID"""
query = """
query GetIssue($id: String!) {
issue(id: $id) {
id
identifier
title
description
priority
priorityLabel
estimate
sortOrder
createdAt
updatedAt
completedAt
canceledAt
dueDate
url
branchName
creator {
id
name
email
displayName
avatarUrl
}
assignee {
id
name
email
displayName
avatarUrl
}
state {
id
name
type
color
position
}
team {
id
key
name
}
project {
id
name
description
state
progress
startDate
targetDate
}
cycle {
id
number
name
description
startsAt
endsAt
completedAt
progress
}
parent {
id
identifier
title
}
labels {
nodes {
id
name
color
description
}
}
attachments {
nodes {
id
title
subtitle
url
metadata
createdAt
}
}
comments {
nodes {
id
body
createdAt
updatedAt
user {
id
name
email
displayName
}
}
}
children {
nodes {
id
identifier
title
state {
id
name
type
}
}
}
relations {
nodes {
id
type
relatedIssue {
id
identifier
title
}
}
}
}
}
"""
variables = {"id": issue_id}
result = await self.execute_query(query, variables)
return cast(dict[str, Any], result["data"]["issue"])

13
toolkits/linear/arcade_linear/constants.py Normal file
View file

@ -0,0 +1,13 @@
import os
LINEAR_API_URL = "https://api.linear.app/graphql"
try:
LINEAR_MAX_CONCURRENT_REQUESTS = int(os.getenv("LINEAR_MAX_CONCURRENT_REQUESTS", 3))
except ValueError:
LINEAR_MAX_CONCURRENT_REQUESTS = 3
try:
LINEAR_MAX_TIMEOUT_SECONDS = int(os.getenv("LINEAR_MAX_TIMEOUT_SECONDS", 30))
except ValueError:
LINEAR_MAX_TIMEOUT_SECONDS = 30

194
toolkits/linear/arcade_linear/models.py Normal file
View file

@ -0,0 +1,194 @@
"""Linear toolkit models and enums"""
from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import Optional
class DateRange(Enum):
"""Date range enum for Linear datetime queries
Provides consistent datetime range handling.
All ranges are calculated in UTC and return timezone-aware datetime objects.
"""
TODAY = "today"
YESTERDAY = "yesterday"
THIS_WEEK = "this_week"
LAST_WEEK = "last_week"
THIS_MONTH = "this_month"
LAST_MONTH = "last_month"
THIS_YEAR = "this_year"
LAST_YEAR = "last_year"
LAST_7_DAYS = "last_7_days"
LAST_30_DAYS = "last_30_days"
def to_datetime_range(
self, reference_time: datetime | None = None
) -> tuple[datetime, datetime]:
"""Convert DateRange to start and end datetime objects
Args:
reference_time: Optional reference time for calculations. Defaults to now in UTC.
Returns:
Tuple of (start_datetime, end_datetime) - both timezone-aware in UTC
"""
now = reference_time or datetime.now(timezone.utc)
# Map enum values to their corresponding helper methods
range_methods = {
DateRange.TODAY: self._get_today_range,
DateRange.YESTERDAY: self._get_yesterday_range,
DateRange.THIS_WEEK: self._get_this_week_range,
DateRange.LAST_WEEK: self._get_last_week_range,
DateRange.THIS_MONTH: self._get_this_month_range,
DateRange.LAST_MONTH: self._get_last_month_range,
DateRange.THIS_YEAR: self._get_this_year_range,
DateRange.LAST_YEAR: self._get_last_year_range,
DateRange.LAST_7_DAYS: lambda n: self._get_last_n_days_range(n, 7),
DateRange.LAST_30_DAYS: lambda n: self._get_last_n_days_range(n, 30),
}
if self in range_methods:
return range_methods[self](now)
else:
raise ValueError("Invalid DateRange enum value")
def _get_today_range(self, now: datetime) -> tuple[datetime, datetime]:
"""Get today's start and end datetime."""
start = now.replace(hour=0, minute=0, second=0, microsecond=0)
end = now.replace(hour=23, minute=59, second=59, microsecond=999999)
return start, end
def _get_yesterday_range(self, now: datetime) -> tuple[datetime, datetime]:
"""Get yesterday's start and end datetime."""
yesterday = now - timedelta(days=1)
start = yesterday.replace(hour=0, minute=0, second=0, microsecond=0)
end = yesterday.replace(hour=23, minute=59, second=59, microsecond=999999)
return start, end
def _get_this_week_range(self, now: datetime) -> tuple[datetime, datetime]:
"""Get this week's start and end datetime."""
# Start of current week (Monday)
start = now - timedelta(days=now.weekday())
start = start.replace(hour=0, minute=0, second=0, microsecond=0)
end = start + timedelta(days=6, hours=23, minutes=59, seconds=59, microseconds=999999)
return start, end
def _get_last_week_range(self, now: datetime) -> tuple[datetime, datetime]:
"""Get last week's start and end datetime."""
# Start of last week (Monday)
start = now - timedelta(days=now.weekday() + 7)
start = start.replace(hour=0, minute=0, second=0, microsecond=0)
end = start + timedelta(days=6, hours=23, minutes=59, seconds=59, microseconds=999999)
return start, end
def _get_this_month_range(self, now: datetime) -> tuple[datetime, datetime]:
"""Get this month's start and end datetime."""
start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
end = now
return start, end
def _get_last_month_range(self, now: datetime) -> tuple[datetime, datetime]:
"""Get last month's start and end datetime."""
# First day of current month
first_of_current_month = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
# Last day of previous month
end = first_of_current_month - timedelta(microseconds=1)
# First day of previous month
start = end.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
return start, end
def _get_this_year_range(self, now: datetime) -> tuple[datetime, datetime]:
"""Get this year's start and end datetime."""
start = now.replace(month=1, day=1, hour=0, minute=0, second=0, microsecond=0)
end = now
return start, end
def _get_last_year_range(self, now: datetime) -> tuple[datetime, datetime]:
"""Get last year's start and end datetime."""
last_year = now.year - 1
start = now.replace(
year=last_year, month=1, day=1, hour=0, minute=0, second=0, microsecond=0
)
end = now.replace(
year=last_year,
month=12,
day=31,
hour=23,
minute=59,
second=59,
microsecond=999999,
)
return start, end
def _get_last_n_days_range(self, now: datetime, days: int) -> tuple[datetime, datetime]:
"""Get range for last N days."""
start = now - timedelta(days=days)
end = now
return start, end
def to_start_datetime(self, reference_time: datetime | None = None) -> datetime:
"""Get just the start datetime for this range
Args:
reference_time: Optional reference time for calculations. Defaults to now in UTC.
Returns:
Start datetime for this range
"""
start, _ = self.to_datetime_range(reference_time)
return start
def to_end_datetime(self, reference_time: datetime | None = None) -> datetime:
"""Get just the end datetime for this range
Args:
reference_time: Optional reference time for calculations. Defaults to now in UTC.
Returns:
End datetime for this range
"""
_, end = self.to_datetime_range(reference_time)
return end
def to_iso_strings(self, reference_time: datetime | None = None) -> tuple[str, str]:
"""Get start and end as ISO format strings
Args:
reference_time: Optional reference time for calculations. Defaults to now in UTC.
Returns:
Tuple of (start_iso, end_iso) strings
"""
start, end = self.to_datetime_range(reference_time)
return start.isoformat(), end.isoformat()
@classmethod
def from_string(cls, date_str: str) -> Optional["DateRange"]:
"""Create DateRange from string if it matches a known value
Args:
date_str: String representation of date range
Returns:
DateRange enum or None if no match found
"""
normalized = date_str.lower().strip()
# Direct mapping
value_map = {
"today": cls.TODAY,
"yesterday": cls.YESTERDAY,
"this week": cls.THIS_WEEK,
"last week": cls.LAST_WEEK,
"this month": cls.THIS_MONTH,
"last month": cls.LAST_MONTH,
"this year": cls.THIS_YEAR,
"last year": cls.LAST_YEAR,
"last 7 days": cls.LAST_7_DAYS,
"last 30 days": cls.LAST_30_DAYS,
}
return value_map.get(normalized)

9
toolkits/linear/arcade_linear/tools/__init__.py Normal file
View file

@ -0,0 +1,9 @@
"""Linear tools for Arcade AI"""
from arcade_linear.tools.issues import get_issue
from arcade_linear.tools.teams import get_teams
__all__ = [
"get_issue",
"get_teams",
]

88
toolkits/linear/arcade_linear/tools/issues.py Normal file
View file

@ -0,0 +1,88 @@
from typing import Annotated, Any
from arcade_tdk import ToolContext, tool
from arcade_tdk.auth import Linear
from arcade_linear.client import LinearClient
from arcade_linear.utils import clean_issue_data, parse_date_string
@tool(requires_auth=Linear(scopes=["read"]))
async def get_issue(
context: ToolContext,
issue_id: Annotated[
str, "The Linear issue ID or identifier (e.g. 'FE-123', 'API-456') to retrieve."
],
include_comments: Annotated[
bool, "Whether to include comments in the response. Defaults to True."
] = True,
include_attachments: Annotated[
bool, "Whether to include attachments in the response. Defaults to True."
] = True,
include_relations: Annotated[
bool,
"Whether to include issue relations (blocks, dependencies) in the response. "
"Defaults to True.",
] = True,
include_children: Annotated[
bool, "Whether to include sub-issues in the response. Defaults to True."
] = True,
) -> Annotated[dict[str, Any], "Complete issue details with related data"]:
"""Get detailed information about a specific Linear issue
This tool retrieves complete information about a single Linear issue when you have
its specific ID or identifier. It's purely for reading and viewing data.
What this tool provides:
- Complete issue details (title, description, status, assignee, etc.)
- Comments and discussion history (if requested)
- File attachments (if requested)
- Related issues and dependencies (if requested)
- Sub-issues and hierarchical relationships (if requested)
When to use this tool:
- When you need to examine the full details of a specific issue
- When you want to read issue content, comments, or relationships
- When you need to analyze or compare issue information
- When you have an issue ID and need to understand its current state
When NOT to use this tool:
- Do NOT use this if you need to change, modify, or update anything
- Do NOT use this if you're trying to create new issues
- Do NOT use this if you're searching for multiple issues
This tool is READ-ONLY - it cannot make any changes to issues.
"""
client = LinearClient(context.get_auth_token_or_empty())
# Get issue data
issue_data = await client.get_issue_by_id(issue_id)
if not issue_data:
return {"error": f"Issue not found: {issue_id}"}
# Clean and format the issue data
cleaned_issue = clean_issue_data(issue_data)
# Optionally remove certain fields based on parameters
if not include_comments:
cleaned_issue.pop("comments", None)
if not include_attachments:
cleaned_issue.pop("attachments", None)
if not include_relations:
cleaned_issue.pop("relations", None)
if not include_children:
cleaned_issue.pop("children", None)
# Get current timestamp for retrieval time
current_time = parse_date_string("now")
retrieved_at = current_time.isoformat() if current_time else None
return {
"issue": cleaned_issue,
"retrieved_at": retrieved_at,
}

107
toolkits/linear/arcade_linear/tools/teams.py Normal file
View file

@ -0,0 +1,107 @@
from typing import Annotated, Any
from arcade_tdk import ToolContext, tool
from arcade_tdk.auth import Linear
from arcade_linear.client import LinearClient
from arcade_linear.utils import (
add_pagination_info,
clean_team_data,
parse_date_string,
validate_date_format,
)
@tool(requires_auth=Linear(scopes=["read"]))
async def get_teams(
context: ToolContext,
team_name: Annotated[
str | None,
"Filter by team name. Provide specific team name (e.g. 'Frontend', 'Product Web') "
"or partial name. Use this to find specific teams or check team membership. "
"Defaults to None (all teams).",
] = None,
include_archived: Annotated[
bool, "Whether to include archived teams in results. Defaults to False."
] = False,
created_after: Annotated[
str | None,
"Filter teams created after this date. Can be:\n"
"- Relative date string (e.g. 'last month', 'this week', 'yesterday')\n"
"- ISO date string (e.g. 'YYYY-MM-DD')\n"
"Defaults to None (all time).",
] = None,
limit: Annotated[
int, "Maximum number of teams to return. Min 1, max 100. Defaults to 50."
] = 50,
end_cursor: Annotated[
str | None,
"Cursor for pagination - get teams after this cursor. Use the 'end_cursor' "
"from previous response. Defaults to None (start from beginning).",
] = None,
) -> Annotated[dict[str, Any], "Teams in the workspace with member information"]:
"""Get Linear teams and team information including team members
This tool retrieves team information from your Linear workspace, including team details,
settings, and member information. Use this tool for team discovery and team membership queries.
What this tool provides:
- Team basic information (name, key, description)
- Team members and their roles
- Team settings and configuration
- Team creation and status information
- Team hierarchy and relationships
This tool is the primary way to get team information.
"""
# Validate inputs
limit = max(1, min(limit, 100))
# Parse and validate date
created_after_date = None
if created_after:
# Validate and parse string (handles DateRange enum strings internally)
validate_date_format("created_after", created_after)
created_after_date = parse_date_string(created_after)
client = LinearClient(context.get_auth_token_or_empty())
# Get teams with filtering
teams_response = await client.get_teams(
first=limit,
after=end_cursor,
include_archived=include_archived,
name_filter=team_name,
)
# Apply additional filtering if needed
teams = teams_response["nodes"]
# Filter by creation date if specified
if created_after_date:
filtered_teams = []
for team in teams:
team_created_at = parse_date_string(team.get("createdAt", ""))
if team_created_at and team_created_at >= created_after_date:
filtered_teams.append(team)
teams = filtered_teams
# Clean and format teams
cleaned_teams = [clean_team_data(team) for team in teams]
response = {
"teams": cleaned_teams,
"total_count": len(cleaned_teams),
"filters": {
"team_name": team_name,
"include_archived": include_archived,
"created_after": created_after,
},
}
# Add pagination info
if "pageInfo" in teams_response and teams_response["pageInfo"].get("hasNextPage"):
add_pagination_info(response, teams_response["pageInfo"])
return response

332
toolkits/linear/arcade_linear/utils.py Normal file
View file

@ -0,0 +1,332 @@
from datetime import datetime, timezone
from typing import Any, cast
import dateparser
from arcade_tdk.errors import ToolExecutionError
from arcade_linear.models import DateRange
# Error message constants
INVALID_DATE_FORMAT_ERROR = "Invalid date format for {field}: '{value}'"
def remove_none_values(data: dict[str, Any]) -> dict[str, Any]:
"""Remove None values from a dictionary"""
return {k: v for k, v in data.items() if v is not None}
def parse_date_string(date_str: str) -> datetime | None:
"""Parse a date string into a timezone-aware datetime object
First tries to match against DateRange enum for consistent relative dates,
then falls back to dateparser for flexible parsing.
"""
if not date_str:
return None
# Check if it's a relative time expression that matches our DateRange enum
date_range = DateRange.from_string(date_str)
if date_range:
# For relative dates, return the start of the range
# This maintains backward compatibility with existing usage
return date_range.to_start_datetime()
# Fall back to dateparser for other formats (ISO dates, etc.)
try:
parsed_date = dateparser.parse(date_str)
if parsed_date is None:
return None
# Cast to datetime to satisfy type checker
parsed_datetime = cast(datetime, parsed_date)
# Ensure all datetimes are timezone-aware (UTC)
if parsed_datetime.tzinfo is None:
return parsed_datetime.replace(tzinfo=timezone.utc)
else:
return parsed_datetime
except Exception:
return None
def parse_date_range(date_str: str) -> tuple[datetime, datetime] | None:
"""Parse a date string into a datetime range tuple if it matches a DateRange enum
Args:
date_str: String that might represent a date range
Returns:
Tuple of (start_datetime, end_datetime) or None if not a valid range
"""
if not date_str:
return None
date_range = DateRange.from_string(date_str)
if date_range:
return date_range.to_datetime_range()
return None
def validate_date_format(field_name: str, date_str: str | None) -> None:
"""Validate date format and raise error if invalid"""
if not date_str:
return
parsed_date = parse_date_string(date_str)
if parsed_date is None:
raise ToolExecutionError(INVALID_DATE_FORMAT_ERROR.format(field=field_name, value=date_str))
def clean_user_data(user_data: dict[str, Any]) -> dict[str, Any]:
"""Clean and format user data"""
if not user_data:
return {}
return remove_none_values({
"id": user_data.get("id"),
"name": user_data.get("name"),
"email": user_data.get("email"),
"display_name": user_data.get("displayName"),
"avatar_url": user_data.get("avatarUrl"),
})
def clean_team_data(team_data: dict[str, Any]) -> dict[str, Any]:
"""Clean and format team data"""
if not team_data:
return {}
cleaned = {
"id": team_data.get("id"),
"key": team_data.get("key"),
"name": team_data.get("name"),
"description": team_data.get("description"),
"private": team_data.get("private"),
"archived_at": team_data.get("archivedAt"),
"created_at": team_data.get("createdAt"),
"updated_at": team_data.get("updatedAt"),
"icon": team_data.get("icon"),
"color": team_data.get("color"),
}
if team_data.get("members") and team_data["members"].get("nodes"):
cleaned["members"] = [clean_user_data(member) for member in team_data["members"]["nodes"]]
return remove_none_values(cleaned)
def clean_state_data(state_data: dict[str, Any]) -> dict[str, Any]:
"""Clean and format workflow state data"""
if not state_data:
return {}
return remove_none_values({
"id": state_data.get("id"),
"name": state_data.get("name"),
"type": state_data.get("type"),
"color": state_data.get("color"),
"position": state_data.get("position"),
})
def clean_project_data(project_data: dict[str, Any]) -> dict[str, Any]:
"""Clean and format project data"""
if not project_data:
return {}
return remove_none_values({
"id": project_data.get("id"),
"name": project_data.get("name"),
"description": project_data.get("description"),
"state": project_data.get("state"),
"progress": project_data.get("progress"),
"start_date": project_data.get("startDate"),
"target_date": project_data.get("targetDate"),
"url": project_data.get("url"),
})
def clean_label_data(label_data: dict[str, Any]) -> dict[str, Any]:
"""Clean and format label data"""
if not label_data:
return {}
return remove_none_values({
"id": label_data.get("id"),
"name": label_data.get("name"),
"color": label_data.get("color"),
"description": label_data.get("description"),
})
def clean_relation_data(relation_data: dict[str, Any]) -> dict[str, Any]:
"""Clean and format issue relation data"""
if not relation_data:
return {}
cleaned = {
"id": relation_data.get("id"),
"type": relation_data.get("type"),
}
# Clean related issue data
if relation_data.get("relatedIssue"):
cleaned["related_issue"] = {
"id": relation_data["relatedIssue"].get("id"),
"identifier": relation_data["relatedIssue"].get("identifier"),
"title": relation_data["relatedIssue"].get("title"),
}
return remove_none_values(cleaned)
def clean_comment_data(comment_data: dict[str, Any]) -> dict[str, Any]:
"""Clean and format comment data"""
if not comment_data:
return {}
cleaned = {
"id": comment_data.get("id"),
"body": comment_data.get("body"),
"created_at": comment_data.get("createdAt"),
"updated_at": comment_data.get("updatedAt"),
}
# Clean user data for comment author
if comment_data.get("user"):
cleaned["user"] = clean_user_data(comment_data["user"])
return remove_none_values(cleaned)
def clean_attachment_data(attachment_data: dict[str, Any]) -> dict[str, Any]:
"""Clean and format attachment data"""
if not attachment_data:
return {}
return remove_none_values({
"id": attachment_data.get("id"),
"title": attachment_data.get("title"),
"subtitle": attachment_data.get("subtitle"),
"url": attachment_data.get("url"),
"metadata": attachment_data.get("metadata"),
"created_at": attachment_data.get("createdAt"),
})
def _clean_issue_relations(issue_data: dict[str, Any]) -> list[dict[str, Any]]:
"""Clean issue relations data"""
relations = issue_data.get("relations", {})
if not relations or not relations.get("nodes"):
return []
cleaned_relations = []
for relation in relations["nodes"]:
if relation and relation.get("relatedIssue"):
cleaned_relations.append({
"id": relation.get("id"),
"type": relation.get("type"),
"related_issue": {
"id": relation["relatedIssue"].get("id"),
"identifier": relation["relatedIssue"].get("identifier"),
"title": relation["relatedIssue"].get("title"),
},
})
return cleaned_relations
def _clean_issue_children(issue_data: dict[str, Any]) -> list[dict[str, Any]]:
"""Clean issue children data"""
children = issue_data.get("children", {})
if not children or not children.get("nodes"):
return []
cleaned_children = []
for child in children["nodes"]:
if child:
cleaned_children.append({
"id": child.get("id"),
"identifier": child.get("identifier"),
"title": child.get("title"),
"state": clean_state_data(child.get("state", {})),
})
return cleaned_children
def _clean_issue_labels(issue_data: dict[str, Any]) -> list[dict[str, Any]]:
"""Clean issue labels data"""
labels = issue_data.get("labels", {})
if not labels or not labels.get("nodes"):
return []
return [clean_label_data(label) for label in labels["nodes"] if label]
def _clean_issue_comments(issue_data: dict[str, Any]) -> list[dict[str, Any]]:
"""Clean issue comments data"""
comments = issue_data.get("comments", {})
if not comments or not comments.get("nodes"):
return []
return [clean_comment_data(comment) for comment in comments["nodes"] if comment]
def _clean_issue_attachments(issue_data: dict[str, Any]) -> list[dict[str, Any]]:
"""Clean issue attachments data"""
attachments = issue_data.get("attachments", {})
if not attachments or not attachments.get("nodes"):
return []
return [clean_attachment_data(attachment) for attachment in attachments["nodes"] if attachment]
def clean_issue_data(issue_data: dict[str, Any]) -> dict[str, Any]:
"""Clean and format issue data for consistent output"""
if not issue_data:
return {}
# Clean basic issue data
cleaned_issue = {
"id": issue_data.get("id"),
"identifier": issue_data.get("identifier"),
"title": issue_data.get("title"),
"description": issue_data.get("description"),
"priority": issue_data.get("priority"),
"priority_label": issue_data.get("priorityLabel"),
"estimate": issue_data.get("estimate"),
"sort_order": issue_data.get("sortOrder"),
"created_at": issue_data.get("createdAt"),
"updated_at": issue_data.get("updatedAt"),
"completed_at": issue_data.get("completedAt"),
"canceled_at": issue_data.get("canceledAt"),
"due_date": issue_data.get("dueDate"),
"url": issue_data.get("url"),
"branch_name": issue_data.get("branchName"),
"creator": clean_user_data(issue_data.get("creator", {})),
"assignee": clean_user_data(issue_data.get("assignee", {})),
"state": clean_state_data(issue_data.get("state", {})),
"team": clean_team_data(issue_data.get("team", {})),
"project": clean_project_data(issue_data.get("project", {})),
"parent": clean_issue_data(issue_data.get("parent", {}))
if issue_data.get("parent")
else None,
"labels": _clean_issue_labels(issue_data),
"comments": _clean_issue_comments(issue_data),
"attachments": _clean_issue_attachments(issue_data),
"relations": _clean_issue_relations(issue_data),
"children": _clean_issue_children(issue_data),
}
return remove_none_values(cleaned_issue)
def add_pagination_info(response: dict[str, Any], page_info: dict[str, Any]) -> dict[str, Any]:
"""Add pagination information to response"""
response["pagination"] = {
"has_next_page": page_info.get("hasNextPage", False),
"has_previous_page": page_info.get("hasPreviousPage", False),
"start_cursor": page_info.get("startCursor"),
"end_cursor": page_info.get("endCursor"),
}
return response

335
toolkits/linear/conftest.py Normal file
View file

@ -0,0 +1,335 @@
import json
import random
import string
from collections.abc import Callable
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
from arcade_tdk import ToolAuthorizationContext, ToolContext
# Seed random generator for deterministic tests
random.seed(42)
# Hardcoded email list for deterministic testing with varied domains
TEST_EMAILS = [
"alice.smith@testcorp.com",
"bob.jones@acme.org",
"charlie.brown@techstart.io",
"diana.wilson@example.net",
"eve.davis@startup.co",
"frank.miller@bigtech.com",
"grace.taylor@innovation.ai",
"henry.anderson@devteam.dev",
"iris.johnson@design.studio",
"jack.white@cloudops.tech",
"karen.thomas@product.team",
"liam.jackson@engineering.co",
"mia.harris@marketing.agency",
"noah.martin@sales.pro",
"olivia.garcia@support.help",
"peter.rodriguez@finance.biz",
"quinn.lewis@legal.firm",
"rachel.lee@hr.people",
"sam.walker@operations.work",
"tina.hall@consulting.group",
]
_email_counter = 0
@pytest.fixture
def fake_auth_token() -> str:
return generate_random_str()
def generate_random_str(length: int = 8) -> str:
"""Generate a deterministic random string for testing"""
return "".join(random.choice(string.ascii_letters + string.digits) for _ in range(length)) # noqa: S311
def generate_random_int(min_val: int = 1, max_val: int = 9999) -> int:
"""Generate a deterministic random integer for testing"""
return random.randint(min_val, max_val) # noqa: S311
def get_test_email() -> str:
"""Get the next email from the hardcoded list, cycling through them"""
global _email_counter
email = TEST_EMAILS[_email_counter % len(TEST_EMAILS)]
_email_counter += 1
return email
@pytest.fixture
def generate_random_email() -> Callable[[str | None, str | None], str]:
def random_email_generator(name: str | None = None, domain: str | None = None) -> str:
# If specific name/domain provided, use them, otherwise use hardcoded emails
if name is None and domain is None:
return get_test_email()
name = name or generate_random_str()
domain = domain or "example.com"
return f"{name}@{domain}"
return random_email_generator
@pytest.fixture
def mock_context(fake_auth_token: str) -> ToolContext:
mock_auth = ToolAuthorizationContext(token=fake_auth_token)
return ToolContext(authorization=mock_auth)
@pytest.fixture
def mock_httpx_client():
"""Mock httpx.AsyncClient for GraphQL requests"""
with patch("arcade_linear.client.httpx.AsyncClient") as mock_client_class:
# Create an async mock for the client instance
mock_client_instance = MagicMock()
# Mock the async context manager methods
mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client_instance)
mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None)
# Make the post method async
mock_client_instance.post = AsyncMock()
yield mock_client_instance
@pytest.fixture
def mock_httpx_response() -> Callable[[int, dict], httpx.Response]:
"""Create mock httpx.Response objects"""
def generate_mock_httpx_response(status_code: int, json_data: dict) -> httpx.Response:
response = MagicMock(spec=httpx.Response)
response.status_code = status_code
response.json.return_value = json_data
response.reason_phrase = "OK" if status_code == 200 else "Error"
response.text = json.dumps(json_data)
return response
return generate_mock_httpx_response
# Linear-specific test data builders
@pytest.fixture
def build_user_dict(
generate_random_email: Callable[[str | None, str | None], str],
) -> Callable:
def user_dict_builder(
id_: str | None = None,
email: str | None = None,
name: str | None = None,
display_name: str | None = None,
active: bool = True,
) -> dict[str, Any]:
name = name or generate_random_str()
return {
"id": id_ or generate_random_str(),
"name": name,
"email": email or generate_random_email(name=name),
"displayName": display_name or name,
"avatarUrl": f"https://avatar.example.com/{generate_random_str()}.png",
"active": active,
}
return user_dict_builder
@pytest.fixture
def build_team_dict() -> Callable:
def team_dict_builder(
id_: str | None = None,
key: str | None = None,
name: str | None = None,
description: str | None = None,
) -> dict[str, Any]:
name = name or generate_random_str()
return {
"id": id_ or generate_random_str(),
"key": key or generate_random_str(3).upper(),
"name": name,
"description": description or f"Description for {name}",
"private": False,
"archivedAt": None,
"createdAt": "2023-01-01T00:00:00.000Z",
"updatedAt": "2023-01-01T00:00:00.000Z",
"icon": "🚀",
"color": "#FF6B6B",
"cyclesEnabled": True,
"issueEstimationType": "exponential",
"organization": {"id": generate_random_str(), "name": "Test Organization"},
"members": {"nodes": []},
}
return team_dict_builder
@pytest.fixture
def build_issue_dict(build_user_dict: Callable, build_team_dict: Callable) -> Callable:
def issue_dict_builder(
id_: str | None = None,
identifier: str | None = None,
title: str | None = None,
description: str | None = None,
priority: int = 2,
priority_label: str = "Medium",
) -> dict[str, Any]:
user = build_user_dict()
team = build_team_dict()
return {
"id": id_ or generate_random_str(),
"identifier": identifier or f"TEST-{generate_random_int(1, 9999)}",
"title": title or f"Test Issue {generate_random_str()}",
"description": description or f"Description for test issue {generate_random_str()}",
"priority": priority,
"priorityLabel": priority_label,
"estimate": None,
"sortOrder": 100.0,
"createdAt": "2023-01-01T00:00:00.000Z",
"updatedAt": "2023-01-01T00:00:00.000Z",
"completedAt": None,
"canceledAt": None,
"dueDate": None,
"url": f"https://linear.app/test/issue/{identifier or 'TEST-1'}",
"branchName": None,
"creator": user,
"assignee": user,
"state": {
"id": generate_random_str(),
"name": "Todo",
"type": "unstarted",
"color": "#e2e2e2",
"position": 1,
},
"team": team,
"project": None,
"cycle": None,
"parent": None,
"labels": {"nodes": []},
"children": {"nodes": []},
"relations": {"nodes": []},
}
return issue_dict_builder
@pytest.fixture
def build_workflow_state_dict(build_team_dict: Callable) -> Callable:
def workflow_state_dict_builder(
id_: str | None = None,
name: str | None = None,
type_: str = "unstarted",
color: str = "#e2e2e2",
position: float = 1.0,
) -> dict[str, Any]:
team = build_team_dict()
return {
"id": id_ or generate_random_str(),
"name": name or f"State {generate_random_str()}",
"description": f"Description for {name or 'test state'}",
"type": type_,
"color": color,
"position": position,
"team": team,
}
return workflow_state_dict_builder
@pytest.fixture
def build_cycle_dict(build_team_dict: Callable) -> Callable:
def cycle_dict_builder(
id_: str | None = None,
number: int | None = None,
name: str | None = None,
description: str | None = None,
) -> dict[str, Any]:
team = build_team_dict()
number = number or generate_random_int(1, 100)
return {
"id": id_ or generate_random_str(),
"number": number,
"name": name or f"Sprint {number}",
"description": description or f"Description for Sprint {number}",
"startsAt": "2023-01-01T00:00:00.000Z",
"endsAt": "2023-01-14T23:59:59.000Z",
"completedAt": None,
"autoArchivedAt": None,
"progress": 0.5,
"createdAt": "2023-01-01T00:00:00.000Z",
"updatedAt": "2023-01-01T00:00:00.000Z",
"team": team,
"issues": {"nodes": []},
}
return cycle_dict_builder
@pytest.fixture
def build_project_dict(build_user_dict: Callable) -> Callable:
def project_dict_builder(
id_: str | None = None,
name: str | None = None,
description: str | None = None,
state: str = "planned",
) -> dict[str, Any]:
user = build_user_dict()
return {
"id": id_ or generate_random_str(),
"name": name or f"Project {generate_random_str()}",
"description": description or "Description for test project",
"state": state,
"progress": 0.3,
"startDate": "2023-01-01",
"targetDate": "2023-12-31",
"completedAt": None,
"canceledAt": None,
"autoArchivedAt": None,
"createdAt": "2023-01-01T00:00:00.000Z",
"updatedAt": "2023-01-01T00:00:00.000Z",
"icon": "📋",
"color": "#4F46E5",
"creator": user,
"lead": user,
"teams": {"nodes": []},
"members": {"nodes": []},
}
return project_dict_builder
# GraphQL response builders
@pytest.fixture
def build_graphql_response() -> Callable[[dict], dict]:
def graphql_response_builder(data: dict, errors: list | None = None) -> dict:
response = {"data": data}
if errors:
response["errors"] = errors
return response
return graphql_response_builder
@pytest.fixture
def build_paginated_response() -> Callable[[list, bool, str | None, str | None], dict]:
def paginated_response_builder(
nodes: list,
has_next_page: bool = False,
start_cursor: str | None = None,
end_cursor: str | None = None,
) -> dict:
return {
"nodes": nodes,
"pageInfo": {
"hasNextPage": has_next_page,
"hasPreviousPage": False,
"startCursor": start_cursor,
"endCursor": end_cursor,
},
}
return paginated_response_builder

103
toolkits/linear/evals/eval_issues.py Normal file
View file

@ -0,0 +1,103 @@
"""
Evaluation suite for Linear get_issue tool.
"""
from arcade_evals import (
BinaryCritic,
EvalRubric,
EvalSuite,
ExpectedToolCall,
tool_eval,
)
from arcade_tdk import ToolCatalog
import arcade_linear
from arcade_linear.tools.issues import get_issue
# Evaluation rubric
rubric = EvalRubric(
fail_threshold=0.85,
warn_threshold=0.95,
)
# Tool catalog
catalog = ToolCatalog()
catalog.add_module(arcade_linear)
@tool_eval()
def get_issue_eval_suite() -> EvalSuite:
"""Comprehensive evaluation suite for get_issue tool"""
suite = EvalSuite(
name="Get Issue Evaluation",
system_message=(
"You are an AI assistant with access to Linear tools. "
"Use them to help the user get detailed information about Linear issues."
),
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Get complete issue details",
user_message="Show me complete details for issue API-789",
expected_tool_calls=[
ExpectedToolCall(
func=get_issue,
args={
"issue_id": "API-789",
"include_comments": True,
"include_attachments": True,
"include_relations": True,
"include_children": True,
},
),
],
critics=[
BinaryCritic(critic_field="issue_id", weight=0.6),
BinaryCritic(critic_field="include_comments", weight=0.1),
BinaryCritic(critic_field="include_attachments", weight=0.1),
BinaryCritic(critic_field="include_relations", weight=0.1),
BinaryCritic(critic_field="include_children", weight=0.1),
],
)
suite.add_case(
name="Get issue dependencies",
user_message="Find all dependencies for issue PROJ-100",
expected_tool_calls=[
ExpectedToolCall(
func=get_issue,
args={
"issue_id": "PROJ-100",
"include_relations": True,
},
),
],
critics=[
BinaryCritic(critic_field="issue_id", weight=0.7),
BinaryCritic(critic_field="include_relations", weight=0.3),
],
)
suite.add_case(
name="Get issue with sub-issues and dependencies",
user_message="Get issue FE-123 with all related sub-issues and dependencies",
expected_tool_calls=[
ExpectedToolCall(
func=get_issue,
args={
"issue_id": "FE-123",
"include_relations": True,
"include_children": True,
},
),
],
critics=[
BinaryCritic(critic_field="issue_id", weight=0.5),
BinaryCritic(critic_field="include_relations", weight=0.25),
BinaryCritic(critic_field="include_children", weight=0.25),
],
)
return suite

175
toolkits/linear/evals/eval_teams.py Normal file
View file

@ -0,0 +1,175 @@
"""
Evaluation suite for Linear get_teams tool.
"""
from arcade_evals import (
BinaryCritic,
EvalRubric,
EvalSuite,
ExpectedToolCall,
SimilarityCritic,
tool_eval,
)
from arcade_tdk import ToolCatalog
import arcade_linear
from arcade_linear.tools.teams import get_teams
# Evaluation rubric
rubric = EvalRubric(
fail_threshold=0.85,
warn_threshold=0.95,
)
catalog = ToolCatalog()
catalog.add_module(arcade_linear)
@tool_eval()
def teams_eval_suite() -> EvalSuite:
"""Comprehensive evaluation suite for get_teams tool"""
suite = EvalSuite(
name="Teams Management Evaluation",
system_message=(
"You are an AI assistant with access to Linear tools. "
"Use them to help the user manage Linear teams and organizational structure."
),
catalog=catalog,
rubric=rubric,
)
suite.add_case(
name="Get all teams in workspace",
user_message="Show me all teams in our workspace",
expected_tool_calls=[
ExpectedToolCall(
func=get_teams,
args={},
),
],
critics=[], # No specific args expected
)
suite.add_case(
name="Find recently created teams",
user_message="Which teams were created in the last month?",
expected_tool_calls=[
ExpectedToolCall(
func=get_teams,
args={
"created_after": "last month",
},
),
],
critics=[
BinaryCritic(critic_field="created_after", weight=1.0),
],
)
suite.add_case(
name="Find teams created this week",
user_message="Show me teams created this week",
expected_tool_calls=[
ExpectedToolCall(
func=get_teams,
args={
"created_after": "this week",
},
),
],
critics=[
BinaryCritic(critic_field="created_after", weight=1.0),
],
)
suite.add_case(
name="Find teams created in last 7 days",
user_message="Which teams were created in the last 7 days?",
expected_tool_calls=[
ExpectedToolCall(
func=get_teams,
args={
"created_after": "last 7 days",
},
),
],
critics=[
BinaryCritic(critic_field="created_after", weight=1.0),
],
)
suite.add_case(
name="Get active teams only",
user_message="Find teams that aren't archived",
expected_tool_calls=[
ExpectedToolCall(
func=get_teams,
args={
"include_archived": False,
},
),
],
critics=[
BinaryCritic(critic_field="include_archived", weight=1.0),
],
)
suite.add_case(
name="Search teams by name",
user_message='Find teams that have "Engineering" in their name',
expected_tool_calls=[
ExpectedToolCall(
func=get_teams,
args={
"team_name": "Engineering",
},
),
],
critics=[
SimilarityCritic(critic_field="team_name", weight=1.0),
],
)
suite.add_case(
name="Get specific team by name",
user_message="Show me the Frontend team details",
expected_tool_calls=[
ExpectedToolCall(
func=get_teams,
args={
"team_name": "Frontend",
},
),
],
critics=[
BinaryCritic(critic_field="team_name", weight=1.0),
],
)
suite.add_case(
name="Clarify ambiguous team request",
user_message="I need to see the Engineering team info",
additional_messages=[
{
"role": "assistant",
"content": (
"I found multiple teams with 'Engineering' in the name. "
"Could you be more specific about which Engineering team you're looking for?"
),
},
{"role": "user", "content": "I meant the Backend Engineering team specifically"},
],
expected_tool_calls=[
ExpectedToolCall(
func=get_teams,
args={
"team_name": "Backend Engineering",
},
),
],
critics=[
SimilarityCritic(critic_field="team_name", weight=1.0),
],
)
return suite

59
toolkits/linear/pyproject.toml Normal file
View file

@ -0,0 +1,59 @@
[build-system]
requires = [ "hatchling",]
build-backend = "hatchling.build"
[project]
name = "arcade_linear"
version = "0.1.0"
description = "Arcade tools designed for LLMs to interact with Linear"
requires-python = ">=3.10"
dependencies = [
"arcade-tdk>=2.0.0,<3.0.0",
"httpx>=0.27.2,<1.0.0",
"dateparser>=1.1.8,<2.0.0",
]
[[project.authors]]
name = "Arcade"
email = "dev@arcade.dev"
[project.optional-dependencies]
dev = [
"arcade-ai[evals]>=2.0.0,<3.0.0",
"arcade-serve>=2.0.0,<3.0.0",
"pytest>=8.3.0,<8.4.0",
"pytest-cov>=4.0.0,<4.1.0",
"pytest-asyncio>=0.24.0,<0.25.0",
"pytest-mock>=3.11.1,<3.12.0",
"mypy>=1.5.1,<1.6.0",
"pre-commit>=3.4.0,<3.5.0",
"tox>=4.11.1,<4.12.0",
"ruff>=0.7.4,<0.8.0",
"types-dateparser>=1.2.2.20250627",
]
# Use local path sources for arcade libs when working locally
[tool.uv.sources]
arcade-ai = {path = "../../", editable = true}
arcade-tdk = { path = "../../libs/arcade-tdk/", editable = true }
arcade-serve = { path = "../../libs/arcade-serve/", editable = true }
[tool.mypy]
files = [ "arcade_linear/**/*.py",]
python_version = "3.10"
disallow_untyped_defs = "True"
disallow_any_unimported = "True"
no_implicit_optional = "True"
check_untyped_defs = "True"
warn_return_any = "True"
warn_unused_ignores = "True"
show_error_codes = "True"
ignore_missing_imports = "True"
[tool.pytest.ini_options]
testpaths = [ "tests",]
[tool.coverage.report]
skip_empty = true
[tool.hatch.build.targets.wheel]
packages = [ "arcade_linear",]

0
toolkits/linear/tests/__init__.py Normal file
View file

154
toolkits/linear/tests/test_client.py Normal file
View file

@ -0,0 +1,154 @@
from unittest.mock import AsyncMock, patch
import pytest
from arcade_tdk.errors import ToolExecutionError
from arcade_linear.client import LinearClient
class TestLinearClient:
"""Tests for LinearClient"""
@pytest.mark.asyncio
async def test_client_init(self):
"""Test client initialization"""
test_token = "test_token" # noqa: S105
client = LinearClient(test_token)
assert client.auth_token == test_token
assert client.api_url == "https://api.linear.app/graphql"
assert client.max_concurrent_requests == 3
assert client.timeout_seconds == 30
def test_build_headers(self):
"""Test header building"""
client = LinearClient("test_token")
headers = client._build_headers()
assert headers["Authorization"] == "Bearer test_token"
assert headers["Content-Type"] == "application/json"
assert headers["Accept"] == "application/json"
def test_build_error_message_graphql_single(self):
"""Test error message building for single GraphQL error"""
client = LinearClient("test_token")
# Mock response with single GraphQL error
response = AsyncMock()
response.status_code = 200
response.json = lambda: {
"errors": [{"message": "Field not found", "extensions": {"code": "FIELD_ERROR"}}]
}
user_msg, dev_msg = client._build_error_message(response)
assert user_msg == "Field not found"
assert "Field not found" in dev_msg
assert "FIELD_ERROR" in dev_msg
def test_build_error_message_http_error(self):
"""Test error message building for HTTP errors"""
client = LinearClient("test_token")
# Mock HTTP error response with valid JSON but no GraphQL errors
response = AsyncMock()
response.status_code = 401
response.reason_phrase = "Unauthorized"
response.text = "Authentication required"
# Return valid JSON that doesn't have "errors" field
response.json = lambda: {"message": "Authentication required"}
user_msg, dev_msg = client._build_error_message(response)
assert user_msg == "HTTP 401: Unauthorized"
assert "HTTP 401: Authentication required" in dev_msg
@pytest.mark.asyncio
async def test_raise_for_status_success(self):
"""Test _raise_for_status with successful response"""
client = LinearClient("test_token")
# Mock successful response
response = AsyncMock()
response.status_code = 200
response.json = lambda: {"data": {"viewer": {"id": "user_1"}}}
# Should not raise exception
client._raise_for_status(response)
@pytest.mark.asyncio
async def test_raise_for_status_graphql_error(self):
"""Test _raise_for_status with GraphQL errors"""
client = LinearClient("test_token")
# Mock response with GraphQL errors
response = AsyncMock()
response.status_code = 200
response.json = lambda: {"errors": [{"message": "Invalid field"}]}
with pytest.raises(ToolExecutionError) as exc_info:
client._raise_for_status(response)
assert "Invalid field" in str(exc_info.value)
@pytest.mark.asyncio
@patch("httpx.AsyncClient")
async def test_execute_query_success(self, mock_http_client):
"""Test successful GraphQL query execution"""
client = LinearClient("test_token")
# Mock HTTP client and response
mock_client_instance = AsyncMock()
mock_http_client.return_value.__aenter__.return_value = mock_client_instance
mock_response = AsyncMock()
mock_response.status_code = 200
mock_response.json = lambda: {"data": {"viewer": {"id": "user_1"}}}
mock_client_instance.post.return_value = mock_response
query = "query { viewer { id } }"
variables = {"test": "value"}
result = await client.execute_query(query, variables)
assert result["data"]["viewer"]["id"] == "user_1"
mock_client_instance.post.assert_called_once()
@pytest.mark.asyncio
@patch("arcade_linear.client.LinearClient.execute_query")
async def test_get_teams(self, mock_execute_query):
"""Test get_teams method"""
client = LinearClient("test_token")
mock_execute_query.return_value = {
"data": {
"teams": {
"nodes": [{"id": "team_1", "name": "Frontend", "key": "FE"}],
"pageInfo": {"hasNextPage": False},
}
}
}
result = await client.get_teams(first=10, include_archived=True)
assert len(result["nodes"]) == 1
assert result["nodes"][0]["name"] == "Frontend"
assert result["pageInfo"]["hasNextPage"] is False
mock_execute_query.assert_called_once()
@pytest.mark.asyncio
@patch("arcade_linear.client.LinearClient.execute_query")
async def test_get_issue_by_id(self, mock_execute_query):
"""Test get_issue_by_id method"""
client = LinearClient("test_token")
mock_execute_query.return_value = {
"data": {"issue": {"id": "issue_1", "identifier": "FE-123", "title": "Test issue"}}
}
result = await client.get_issue_by_id("FE-123")
assert result["id"] == "issue_1"
assert result["identifier"] == "FE-123"
assert result["title"] == "Test issue"
mock_execute_query.assert_called_once()

94
toolkits/linear/tests/test_issues.py Normal file
View file

@ -0,0 +1,94 @@
from unittest.mock import AsyncMock, patch
import pytest
from arcade_linear.tools.issues import get_issue
@pytest.fixture
def mock_context():
"""Mock context for testing"""
context = AsyncMock()
context.get_auth_token_or_empty.return_value = "test-token"
return context
class TestGetIssue:
"""Tests for get_issue tool"""
@pytest.mark.asyncio
@patch("arcade_linear.tools.issues.LinearClient")
async def test_get_issue_success(self, mock_client_class, mock_context):
"""Test successful issue retrieval"""
# Setup mock
mock_client = AsyncMock()
mock_client_class.return_value = mock_client
mock_client.get_issue_by_id.return_value = {
"id": "issue_1",
"identifier": "FE-123",
"title": "Fix authentication bug",
"description": "Authentication not working",
"priority": 1,
"priorityLabel": "Urgent",
"createdAt": "2024-01-01T00:00:00Z",
"team": {"id": "team_1", "key": "FE", "name": "Frontend"},
"assignee": {"id": "user_1", "name": "John Doe"},
"state": {"id": "state_1", "name": "In Progress"},
"labels": {"nodes": []},
"attachments": {"nodes": []},
"comments": {"nodes": []},
"children": {"nodes": []},
"relations": {"nodes": []},
}
# Call function
result = await get_issue(mock_context, "FE-123")
# Assertions
assert "issue" in result
assert result["issue"]["identifier"] == "FE-123"
assert result["issue"]["title"] == "Fix authentication bug"
mock_client.get_issue_by_id.assert_called_once_with("FE-123")
@pytest.mark.asyncio
@patch("arcade_linear.tools.issues.LinearClient")
async def test_get_issue_not_found(self, mock_client_class, mock_context):
"""Test issue retrieval when issue not found"""
# Setup mock
mock_client = AsyncMock()
mock_client_class.return_value = mock_client
mock_client.get_issue_by_id.return_value = None
# Call function
result = await get_issue(mock_context, "NON-EXISTENT")
# Assertions
assert "error" in result
assert "Issue not found" in result["error"]
@pytest.mark.asyncio
@patch("arcade_linear.tools.issues.LinearClient")
async def test_get_issue_selective_includes(self, mock_client_class, mock_context):
"""Test issue retrieval with selective includes"""
# Setup mock
mock_client = AsyncMock()
mock_client_class.return_value = mock_client
mock_client.get_issue_by_id.return_value = {
"id": "issue_1",
"identifier": "FE-123",
"title": "Fix authentication bug",
"comments": {"nodes": [{"id": "comment_1"}]},
"attachments": {"nodes": [{"id": "attachment_1"}]},
}
# Call function without comments and attachments
result = await get_issue(
mock_context,
"FE-123",
include_comments=False,
include_attachments=False,
)
# Assertions
assert "comments" not in result["issue"]
assert "attachments" not in result["issue"]

145
toolkits/linear/tests/test_teams.py Normal file
View file

@ -0,0 +1,145 @@
import pytest
from arcade_tdk import ToolContext
from arcade_linear.tools.teams import get_teams
@pytest.mark.asyncio
async def test_get_teams_success(
mock_context: ToolContext,
mock_httpx_client,
mock_httpx_response,
build_team_dict,
build_graphql_response,
build_paginated_response,
):
"""Test successful team retrieval"""
# Create sample teams
team1 = build_team_dict(name="Team Alpha", key="ALPHA")
team2 = build_team_dict(name="Team Beta", key="BETA")
# Build paginated response
teams_response = build_paginated_response([team1, team2])
# Build full GraphQL response
graphql_data = build_graphql_response({"teams": teams_response})
# Mock the HTTP response
http_response = mock_httpx_response(200, graphql_data)
mock_httpx_client.post.return_value = http_response
result = await get_teams(context=mock_context)
assert result["total_count"] == 2
assert len(result["teams"]) == 2
assert result["teams"][0]["name"] == "Team Alpha"
assert result["teams"][0]["key"] == "ALPHA"
assert result["teams"][1]["name"] == "Team Beta"
assert result["teams"][1]["key"] == "BETA"
# Verify the request was made correctly
mock_httpx_client.post.assert_called_once()
call_args = mock_httpx_client.post.call_args
# Check URL
assert call_args[0][0] == "https://api.linear.app/graphql"
# Check that query is in the request body
request_body = call_args[1]["json"]
assert "query" in request_body
assert "teams" in request_body["query"]
@pytest.mark.asyncio
async def test_get_teams_with_team_filter(
mock_context: ToolContext,
mock_httpx_client,
mock_httpx_response,
build_team_dict,
build_graphql_response,
build_paginated_response,
):
"""Test team retrieval with name filter"""
# Create sample team
team = build_team_dict(name="Engineering Team", key="ENG")
# Build responses
teams_response = build_paginated_response([team])
graphql_data = build_graphql_response({"teams": teams_response})
http_response = mock_httpx_response(200, graphql_data)
mock_httpx_client.post.return_value = http_response
result = await get_teams(context=mock_context, team_name="Engineering")
assert result["total_count"] == 1
assert len(result["teams"]) == 1
assert result["teams"][0]["name"] == "Engineering Team"
# Verify filter was applied in the request
call_args = mock_httpx_client.post.call_args
request_body = call_args[1]["json"]
assert "variables" in request_body
assert "filter" in request_body["variables"]
@pytest.mark.asyncio
async def test_get_teams_empty_result(
mock_context: ToolContext,
mock_httpx_client,
mock_httpx_response,
build_graphql_response,
build_paginated_response,
):
"""Test handling of empty teams list"""
# Build empty response
teams_response = build_paginated_response([])
graphql_data = build_graphql_response({"teams": teams_response})
http_response = mock_httpx_response(200, graphql_data)
mock_httpx_client.post.return_value = http_response
result = await get_teams(context=mock_context)
assert result["total_count"] == 0
assert len(result["teams"]) == 0
@pytest.mark.asyncio
async def test_get_teams_graphql_error(
mock_context: ToolContext,
mock_httpx_client,
mock_httpx_response,
):
"""Test handling of GraphQL errors"""
# Build error response
error_data = {
"data": None,
"errors": [
{"message": "Authentication required", "extensions": {"code": "UNAUTHENTICATED"}}
],
}
http_response = mock_httpx_response(200, error_data)
mock_httpx_client.post.return_value = http_response
# The tool should raise an exception for GraphQL errors
with pytest.raises(Exception) as exc_info:
await get_teams(context=mock_context)
assert "Authentication required" in str(exc_info.value)
@pytest.mark.asyncio
async def test_get_teams_http_error(
mock_context: ToolContext,
mock_httpx_client,
mock_httpx_response,
):
"""Test handling of HTTP errors"""
# Mock HTTP 401 error
error_response = mock_httpx_response(401, {"error": "Unauthorized"})
mock_httpx_client.post.return_value = error_response
with pytest.raises(Exception) as exc_info:
await get_teams(context=mock_context)
# Should contain HTTP status information
assert "401" in str(exc_info.value) or "Unauthorized" in str(exc_info.value)

236
toolkits/linear/tests/test_utils.py Normal file
View file

@ -0,0 +1,236 @@
import pytest
from arcade_tdk.errors import ToolExecutionError
from arcade_linear.models import DateRange
from arcade_linear.utils import (
add_pagination_info,
clean_issue_data,
clean_team_data,
parse_date_range,
parse_date_string,
remove_none_values,
validate_date_format,
)
class TestDateParsing:
"""Tests for date parsing utilities"""
def test_parse_date_string_valid_iso(self):
"""Test date parsing with valid ISO strings"""
result = parse_date_string("2024-01-01")
assert result is not None
assert result.year == 2024
assert result.month == 1
assert result.day == 1
assert result.tzinfo is not None
def test_parse_date_string_with_time(self):
"""Test date parsing with date and time"""
result = parse_date_string("2024-01-01T12:30:00Z")
assert result is not None
assert result.year == 2024
assert result.hour == 12
assert result.minute == 30
def test_parse_date_string_relative(self):
"""Test date parsing with relative strings"""
result = parse_date_string("today")
assert result is not None
result = parse_date_string("yesterday")
assert result is not None
def test_parse_date_string_time_mappings(self):
"""Test date parsing with DateRange enum expressions"""
result = parse_date_string("this week")
assert result is not None
result = parse_date_string("last month")
assert result is not None
result = parse_date_string("yesterday")
assert result is not None
result = parse_date_string("last 7 days")
assert result is not None
def test_parse_date_string_invalid(self):
"""Test date parsing with invalid strings"""
result = parse_date_string("invalid-date")
assert result is None
result = parse_date_string("not-a-date")
assert result is None
def test_parse_date_string_empty(self):
"""Test date parsing with empty string"""
result = parse_date_string("")
assert result is None
result = parse_date_string(None)
assert result is None
def test_validate_date_format_valid(self):
"""Test date validation with valid format"""
# Should not raise exception
validate_date_format("test_field", "2024-01-01")
validate_date_format("test_field", "today")
validate_date_format("test_field", None)
validate_date_format("test_field", "")
def test_validate_date_format_invalid(self):
"""Test date validation with invalid format"""
with pytest.raises(ToolExecutionError) as exc_info:
validate_date_format("test_field", "invalid-date")
assert "Invalid date format for test_field" in str(exc_info.value)
class TestDateRange:
"""Tests for DateRange enum"""
def test_date_range_from_string_valid(self):
"""Test DateRange.from_string with valid strings"""
assert DateRange.from_string("today") == DateRange.TODAY
assert DateRange.from_string("yesterday") == DateRange.YESTERDAY
assert DateRange.from_string("last week") == DateRange.LAST_WEEK
assert DateRange.from_string("this month") == DateRange.THIS_MONTH
assert DateRange.from_string("last month") == DateRange.LAST_MONTH
assert DateRange.from_string("last 7 days") == DateRange.LAST_7_DAYS
assert DateRange.from_string("last 30 days") == DateRange.LAST_30_DAYS
def test_date_range_from_string_case_insensitive(self):
"""Test DateRange.from_string is case insensitive"""
assert DateRange.from_string("TODAY") == DateRange.TODAY
assert DateRange.from_string("Last Week") == DateRange.LAST_WEEK
assert DateRange.from_string("THIS MONTH") == DateRange.THIS_MONTH
def test_date_range_from_string_invalid(self):
"""Test DateRange.from_string with invalid strings"""
assert DateRange.from_string("invalid") is None
assert DateRange.from_string("next week") is None
assert DateRange.from_string("") is None
def test_date_range_to_datetime_range(self):
"""Test DateRange.to_datetime_range returns proper datetime objects"""
today_range = DateRange.TODAY.to_datetime_range()
assert len(today_range) == 2
start, end = today_range
assert start.tzinfo is not None
assert end.tzinfo is not None
assert start <= end
def test_date_range_to_start_datetime(self):
"""Test DateRange.to_start_datetime returns proper datetime"""
start = DateRange.LAST_WEEK.to_start_datetime()
assert start.tzinfo is not None
def test_date_range_to_end_datetime(self):
"""Test DateRange.to_end_datetime returns proper datetime"""
end = DateRange.LAST_MONTH.to_end_datetime()
assert end.tzinfo is not None
class TestParseDateRange:
"""Tests for parse_date_range function"""
def test_parse_date_range_valid(self):
"""Test parse_date_range with valid DateRange strings"""
result = parse_date_range("today")
assert result is not None
assert len(result) == 2
start, end = result
assert start.tzinfo is not None
assert end.tzinfo is not None
def test_parse_date_range_invalid(self):
"""Test parse_date_range with invalid strings"""
assert parse_date_range("invalid") is None
assert parse_date_range("") is None
assert parse_date_range("2024-01-01") is None # ISO date, not range
def test_parse_date_range_relative_dates(self):
"""Test parse_date_range with various relative date strings"""
ranges = ["yesterday", "last week", "this month", "last month", "last 7 days"]
for range_str in ranges:
result = parse_date_range(range_str)
assert result is not None, f"Failed for {range_str}"
start, end = result
assert start <= end, f"Invalid range for {range_str}"
class TestDataCleaning:
"""Tests for data cleaning functions"""
def test_remove_none_values(self):
"""Test removing None values from dictionary"""
data = {"a": 1, "b": None, "c": "test", "d": None}
result = remove_none_values(data)
assert result == {"a": 1, "c": "test"}
def test_clean_team_data(self):
"""Test team data cleaning"""
team_data = {
"id": "team_1",
"key": "FE",
"name": "Frontend",
"description": "Frontend team",
"private": False,
"archivedAt": None,
"createdAt": "2024-01-01T00:00:00Z",
"members": {"nodes": [{"id": "user_1", "name": "John Doe"}]},
}
result = clean_team_data(team_data)
assert result["id"] == "team_1"
assert result["key"] == "FE"
assert result["name"] == "Frontend"
assert len(result["members"]) == 1
assert result["members"][0]["name"] == "John Doe"
def test_clean_issue_data(self):
"""Test issue data cleaning"""
issue_data = {
"id": "issue_1",
"identifier": "FE-123",
"title": "Test issue",
"description": "Issue description",
"priority": 2,
"priorityLabel": "High",
"createdAt": "2024-01-01T00:00:00Z",
"assignee": {"id": "user_1", "name": "John Doe"},
"state": {"id": "state_1", "name": "In Progress"},
"team": {"id": "team_1", "name": "Frontend"},
"labels": {"nodes": [{"id": "label_1", "name": "bug"}]},
"children": {"nodes": []},
}
result = clean_issue_data(issue_data)
assert result["id"] == "issue_1"
assert result["identifier"] == "FE-123"
assert result["title"] == "Test issue"
assert result["assignee"]["name"] == "John Doe"
assert result["state"]["name"] == "In Progress"
assert result["team"]["name"] == "Frontend"
assert len(result["labels"]) == 1
assert result["labels"][0]["name"] == "bug"
def test_add_pagination_info(self):
"""Test adding pagination information"""
response = {"data": "test"}
page_info = {
"hasNextPage": True,
"hasPreviousPage": False,
"startCursor": "start123",
"endCursor": "end456",
}
result = add_pagination_info(response, page_info)
assert result["pagination"]["has_next_page"] is True
assert result["pagination"]["has_previous_page"] is False
assert result["pagination"]["start_cursor"] == "start123"
assert result["pagination"]["end_cursor"] == "end456"