Linear Integration

# Install Linear MCP server
npm install -g @linear/mcp-server

# Run as HTTP server (not stdio)
LINEAR_API_KEY=lin_api_... linear-mcp-server --http --port 8001

from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP
import httpx
import os

mcp = FastMCP("Linear")

@mcp.tool()
async def search_issues(team_id: str, query: str) -> list[dict]:
    """Search Linear issues for a team."""
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://api.linear.app/graphql",
            headers={
                "Authorization": f"Bearer {os.getenv('LINEAR_API_KEY')}",
                "Content-Type": "application/json",
            },
            json={
                "query": """
                query SearchIssues($teamId: String!, $query: String!) {
                  issues(filter: {team: {key: {eq: $teamId}}, title: {contains: $query}}) {
                    nodes {
                      id
                      identifier
                      title
                      description
                      state { name }
                      assignee { name }
                      url
                    }
                  }
                }
                """,
                "variables": {"teamId": team_id, "query": query},
            },
        )
        data = resp.json()
        return data.get("data", {}).get("issues", {}).get("nodes", [])

app = FastAPI()
app.mount("/mcp", mcp.sse_app())

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8001)

# Linear MCP server base URL
LINEAR_MCP_URL=https://your-linear-mcp-server.railway.app

# Linear personal API key (passed to MCP server via Authorization header)
LINEAR_API_KEY=lin_api_...

async def get_linear_issues(
    self,
    team_id: str,
    query: str,
) -> list[dict]:

[
    {
        "id": "abc123",
        "identifier": "ENG-123",
        "title": "Add user authentication",
        "state": "In Progress",
        "url": "https://linear.app/team/issue/ENG-123",
        "description": "Implement OAuth flow with GitHub...",
        "assignee": "Alice",
    }
]

from app.mcp.client import mcp_client

# During PR review
linear_refs = ["ENG-123", "ENG-456"]  # Extracted from PR body

issues = await mcp_client.get_linear_issues(
    team_id="ENG",
    query=" ".join(linear_refs),
)

for issue in issues:
    print(f"{issue['identifier']}: {issue['title']}")
    print(f"  Status: {issue['state']}")
    print(f"  URL: {issue['url']}")

POST https://linear-mcp-server.example.com/
Content-Type: application/json
Authorization: Bearer lin_api_...

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "search_issues",
    "arguments": {
      "team_id": "ENG",
      "query": "ENG-123 authentication"
    }
  }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "[{\"id\": \"abc123\", \"identifier\": \"ENG-123\", \"title\": \"Add user authentication\", ...}]"
      }
    ]
  }
}

# JSON-RPC 2.0 result shape: {"result": {"content": [...]}}
result = data.get("result", data)
if isinstance(result, dict):
    content = result.get("content", result)
    if isinstance(content, list):
        # Each content item may be {type: "text", text: "<json>"}
        for item in content:
            if isinstance(item, dict) and item.get("type") == "text":
                parsed = json.loads(item["text"])
                # parsed is now the list of issues

_MCP_TIMEOUT = 10.0  # seconds

try:
    async with httpx.AsyncClient(timeout=_MCP_TIMEOUT) as client:
        response = await client.post(...)
        response.raise_for_status()
except httpx.TimeoutException:
    logger.warning(
        f"MCP call timed out: {settings.LINEAR_MCP_URL} tool=search_issues"
    )
    return []  # Empty list - review continues without Linear context
except httpx.HTTPStatusError as exc:
    logger.warning(f"Linear MCP returned HTTP {exc.response.status_code}")
    return []
except Exception as exc:
    logger.warning(f"Linear MCP query failed: {exc}")
    return []

# Common patterns:
# - "Fixes ENG-123"
# - "Closes INFRA-456"
# - "Resolves ENG-123, ENG-124"

import re

LINEAR_ISSUE_PATTERN = re.compile(
    r"\b([A-Z]{2,10}-\d+)\b",  # Team prefix (2-10 uppercase letters) + dash + number
    re.IGNORECASE,
)

def extract_linear_issue_ids(pr_title: str, pr_body: str) -> list[str]:
    text = f"{pr_title} {pr_body}"
    matches = LINEAR_ISSUE_PATTERN.findall(text)
    return list(dict.fromkeys(matches))  # Deduplicate while preserving order

# Example:
pr_title = "Fixes ENG-123: Add authentication"
pr_body = "This PR implements ENG-123 and partially addresses ENG-124."
issue_ids = extract_linear_issue_ids(pr_title, pr_body)
# Returns: ["ENG-123", "ENG-124"]

# Branch names like:
# - eng-123-add-auth
# - feature/ENG-456-fix-bug

def extract_issue_from_branch(branch_name: str) -> str | None:
    match = LINEAR_ISSUE_PATTERN.search(branch_name)
    return match.group(1) if match else None

# 1. Extract Linear issue IDs from PR metadata
issue_ids = extract_linear_issue_ids(pr_data["title"], pr_data["body"])
if pr_data.get("head", {}).get("ref"):
    branch_issue = extract_issue_from_branch(pr_data["head"]["ref"])
    if branch_issue:
        issue_ids.append(branch_issue)

# 2. Pull issue details from Linear MCP server
linear_issues = []
if issue_ids:
    linear_issues = await mcp_client.get_linear_issues(
        team_id="ENG",  # TODO: Make this configurable per repo
        query=" ".join(issue_ids),
    )

# 3. Format Linear context for AI review prompt
linear_context = ""
if linear_issues:
    linear_context = "\n\nLINEAR CONTEXT:\n"
    for issue in linear_issues:
        linear_context += f"""- Issue {issue['identifier']}: "{issue['title']}"
  Status: {issue['state']}
  URL: {issue['url']}
  Description: {issue.get('description', 'N/A')[:300]}
"""

# 4. Include in AI review
review_prompt = f"""
Review this pull request:

PR Title: {pr_data['title']}
PR Description: {pr_data['body']}

Changed Files: {changed_files}
{linear_context}

Diff:
{pr_diff}

Provide a structured review covering:
1. Does this PR fully address the linked Linear issues?
2. Are there gaps between issue requirements and implementation?
...
"""

# PR Review: Add authentication flow (ENG-123)

## Linked Issues
✅ **ENG-123**: Add user authentication
   - Status: In Progress
   - Assignee: Alice
   - [View in Linear](https://linear.app/team/issue/ENG-123)

## Summary
This PR implements the OAuth flow described in ENG-123. The implementation
aligns well with the issue requirements:
- ✅ GitHub OAuth integration
- ✅ Token encryption at rest
- ⚠️  Issue mentions rate limiting, but I don't see it implemented

## Suggestions
1. **Missing requirement**: ENG-123 specifies rate limiting for failed auth
   attempts. Consider adding a simple in-memory rate limiter.
   
2. **Test coverage**: Issue states this is a critical security feature.
   Add E2E tests for the OAuth callback flow.

## Verdict
APPROVE_WITH_SUGGESTIONS - Strong implementation, but address rate limiting
before merging per issue requirements.

# Add to Installation model
class Installation(Base):
    # ...
    linear_team_id: str | None = None  # e.g., "ENG", "INFRA"

# Or use a mapping file
LINEAR_TEAM_MAP = {
    "nectr-ai/nectr": "ENG",
    "nectr-ai/docs": "DOCS",
    "nectr-ai/infra": "INFRA",
}

def get_team_id(repo_full_name: str) -> str:
    return LINEAR_TEAM_MAP.get(repo_full_name, "ENG")  # Default to ENG

def extract_team_from_issue_ids(issue_ids: list[str]) -> str | None:
    """Extract team ID from issue identifiers.
    
    Examples:
        ["ENG-123"] -> "ENG"
        ["INFRA-456"] -> "INFRA"
    """
    if not issue_ids:
        return None
    # All issue IDs should have same team prefix
    team_id = issue_ids[0].split("-")[0]
    return team_id

HTTP 401 from Linear MCP server