How to Integrate AI Agents with Linear

{
  "query": "mutation IssueCreate($title: String!, $description: String!, $teamId: String!) { issueCreate(input: {title: $title, description: $description, teamId: $teamId}) { issue { id title url } } }",
  "variables": {
    "title": "{{ $json.title }}",
    "description": "{{ $json.description }}",
    "teamId": "your-team-id"
  }
}

pip install langchain langchain-openai requests python-dotenv

import os
import requests
from langchain.tools import tool
from dotenv import load_dotenv

load_dotenv()

LINEAR_API_KEY = os.getenv("LINEAR_API_KEY")
LINEAR_ENDPOINT = "https://api.linear.app/graphql"
HEADERS = {
    "Authorization": LINEAR_API_KEY,
    "Content-Type": "application/json"
}


def run_linear_query(query: str, variables: dict = None) -> dict:
    """Execute a GraphQL query or mutation against Linear API."""
    payload = {"query": query, "variables": variables or {}}
    response = requests.post(LINEAR_ENDPOINT, json=payload, headers=HEADERS)
    response.raise_for_status()
    return response.json()


@tool
def create_linear_issue(title: str, description: str, priority: int = 2,
                        team_key: str = "ENG") -> str:
    """
    Create a Linear issue. Priority: 0=No priority, 1=Urgent, 2=High, 3=Medium, 4=Low.
    team_key is the team identifier (e.g., 'ENG', 'PROD').
    """
    # First get the team ID from the key
    team_query = """
    query GetTeam($key: String!) {
      teams(filter: {key: {eq: $key}}) {
        nodes { id name key }
      }
    }"""
    team_data = run_linear_query(team_query, {"key": team_key})
    teams = team_data.get("data", {}).get("teams", {}).get("nodes", [])
    if not teams:
        return f"Team with key '{team_key}' not found"

    team_id = teams[0]["id"]

    mutation = """
    mutation CreateIssue($title: String!, $description: String!, $teamId: String!, $priority: Int) {
      issueCreate(input: {
        title: $title,
        description: $description,
        teamId: $teamId,
        priority: $priority
      }) {
        success
        issue { id title url priority }
      }
    }"""
    result = run_linear_query(mutation, {
        "title": title,
        "description": description,
        "teamId": team_id,
        "priority": priority
    })
    issue = result.get("data", {}).get("issueCreate", {}).get("issue", {})
    return f"Issue created: {issue.get('title')} — {issue.get('url')}"


@tool
def search_linear_issues(query_text: str, state: str = "all") -> str:
    """Search Linear issues by text. state can be 'triage', 'backlog', 'todo', 'in_progress', or 'all'."""
    search_query = """
    query SearchIssues($term: String!) {
      issueSearch(term: $term, first: 10) {
        nodes {
          id
          title
          state { name }
          priority
          assignee { name }
          url
        }
      }
    }"""
    result = run_linear_query(search_query, {"term": query_text})
    issues = result.get("data", {}).get("issueSearch", {}).get("nodes", [])
    if not issues:
        return f"No issues found matching '{query_text}'"

    formatted = []
    for issue in issues:
        state_name = issue.get("state", {}).get("name", "Unknown")
        assignee = issue.get("assignee", {}).get("name", "Unassigned") if issue.get("assignee") else "Unassigned"
        formatted.append(f"- {issue['title']} | State: {state_name} | Assignee: {assignee} | {issue['url']}")
    return "\n".join(formatted)


@tool
def get_team_backlog_summary(team_key: str = "ENG") -> str:
    """Get a summary of current backlog items for a team."""
    query = """
    query TeamIssues($filter: IssueFilter) {
      issues(filter: $filter, first: 30, orderBy: priority) {
        nodes {
          title
          priority
          state { name }
          estimate
          labels { nodes { name } }
        }
      }
    }"""
    result = run_linear_query(query, {
        "filter": {
            "team": {"key": {"eq": team_key}},
            "state": {"type": {"in": ["triage", "backlog", "unstarted"]}}
        }
    })
    issues = result.get("data", {}).get("issues", {}).get("nodes", [])
    if not issues:
        return f"No backlog issues found for team {team_key}"

    priority_names = {0: "No priority", 1: "Urgent", 2: "High", 3: "Medium", 4: "Low"}
    formatted = [f"Backlog Summary for team {team_key} ({len(issues)} items):"]
    for issue in issues[:15]:
        p = priority_names.get(issue.get("priority", 0), "Unknown")
        formatted.append(f"- [{p}] {issue['title']}")
    return "\n".join(formatted)

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0.2)
tools = [create_linear_issue, search_linear_issues, get_team_backlog_summary]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a technical project manager assisting with Linear project management.

Your responsibilities:
- Create well-structured issues from raw descriptions or bug reports
- Search for existing issues to prevent duplicates before creating new ones
- Provide backlog summaries and sprint planning recommendations
- Classify issues by appropriate priority

When creating issues:
- Write clear, actionable titles (start with verb: "Fix", "Add", "Update", "Remove")
- Include reproduction steps for bugs
- Include acceptance criteria for features
- Set realistic priority based on user impact"""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Example: Convert customer feedback into Linear issue
result = executor.invoke({
    "input": """Customer feedback received:
"The export button in the reports section doesn't work when there are more than 1000 rows.
The page just freezes and nothing gets downloaded."

Create a bug report in Linear for the ENG team."""
})
print(result["output"])

def process_customer_feedback(feedback_items: list[dict]) -> list[str]:
    """Process a batch of customer feedback into Linear issues."""
    created_issues = []

    for feedback in feedback_items:
        result = executor.invoke({
            "input": f"""Process this customer feedback into a Linear issue:

Source: {feedback['source']}
Customer: {feedback['customer_name']} (Plan: {feedback['plan']})
Feedback: {feedback['text']}

1. First search for similar existing issues to avoid duplicates
2. If no duplicate found, create a Linear issue in the PROD team
3. Set priority based on customer plan (Enterprise = High, Pro = Medium, Free = Low)
4. Include the customer's exact quote in the description"""
        })
        created_issues.append(result["output"])

    return created_issues