Linear Integration

docker compose restart backend
# or
uvicorn app.main:app --reload

from app.mcp.client import mcp_client

issues = await mcp_client.get_linear_issues(
    team_id="ENG",
    query="authentication bug"
)
# Returns: [
#   {"id": "ENG-123", "title": "Fix OAuth redirect loop", "state": "In Progress", "url": "..."},
#   {"id": "ENG-145", "title": "Auth token expiration", "state": "Backlog", "url": "..."},
# ]

curl -X POST http://linear-mcp-server:8001/ \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer lin_api_..." \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
      "name": "search_issues",
      "arguments": {"team_id": "ENG", "query": "OAuth"}
    }
  }'

{
  "result": {
    "content": [
      {
        "type": "text",
        "text": "[{\"id\": \"ENG-123\", \"title\": \"Fix OAuth redirect loop\", \"state\": \"In Progress\"}]"
      }
    ]
  }
}

async def get_linear_issues(self, team_id: str, query: str) -> list[dict]:
    """Pull issues / tasks from the Linear MCP server matching *query*.

    Args:
        team_id: Linear team identifier (e.g. "ENG").
        query:   Free-text search query (topic, feature area, keyword).

    Returns:
        List of issue dicts: {id, title, state, url, description}.
        Empty list if Linear MCP is not configured or the call fails.
    """
    if not settings.LINEAR_MCP_URL:
        logger.info(
            "LINEAR_MCP_URL not configured — skipping Linear issue fetch "
            "(set LINEAR_MCP_URL + LINEAR_API_KEY to enable)"
        )
        return []

    return await self.query_mcp_server(
        server_url=settings.LINEAR_MCP_URL,
        tool_name="search_issues",
        args={"team_id": team_id, "query": query},
        auth_token=settings.LINEAR_API_KEY,
    )

async def query_mcp_server(
    self,
    server_url: str,
    tool_name: str,
    args: dict,
    auth_token: str | None = None,
) -> list[dict] | dict:
    """Generic method to call any MCP server tool over HTTP/SSE JSON-RPC."""
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": {"name": tool_name, "arguments": args},
    }
    headers: dict[str, str] = {"Content-Type": "application/json"}
    if auth_token:
        headers["Authorization"] = f"Bearer {auth_token}"

    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.post(
            f"{server_url.rstrip('/')}/",
            json=payload,
            headers=headers,
        )
        response.raise_for_status()
        data = response.json()

    # Unwrap MCP content array: {"result": {"content": [{"type": "text", "text": "..."}]}}
    result = data.get("result", data)
    content = result.get("content", result)
    # ...