✨ refactor: simplify API endpoint detection using utility function

pyproject.toml

@@ -5,7 +5,7 @@ description = "Integrates CodeLogic's powerful codebase knowledge graphs with a
 readme = "README.md"
 license = "MPL-2.0"
 requires-python = ">=3.13"
-dependencies = [ "debugpy>=1.8.12", "httpx>=0.28.1", "mcp[cli]>=1.3.0", "pip-licenses>=5.0.0", "python-dotenv>=1.0.1", "tenacity>=9.0.0",]
+dependencies = [ "debugpy>=1.8.12", "httpx>=0.28.1", "mcp[cli]>=1.3.0", "pip-licenses>=5.0.0", "python-dotenv>=1.0.1", "tenacity>=9.0.0", "toml>=0.10.2",]
 [[project.authors]]
 name = "garrmark"
 email = "[email protected]"

src/codelogic_mcp_server/handlers.py

@@ -17,7 +17,7 @@
 import sys
 from .server import server
 import mcp.types as types
-from .utils import extract_nodes, extract_relationships, get_mv_id, get_method_nodes, get_impact, find_node_by_id, search_database_entity, process_database_entity_impact, generate_combined_database_report
+from .utils import extract_nodes, extract_relationships, get_mv_id, get_method_nodes, get_impact, find_node_by_id, search_database_entity, process_database_entity_impact, generate_combined_database_report, find_api_endpoints
 import time
 from datetime import datetime
 
@@ -227,13 +227,13 @@ async def handle_method_impact(arguments: dict | None) -> list[types.TextContent
     # Extract code owners and reviewers
     code_owners = target_node['properties'].get('codelogic.owners', []) if target_node else []
     code_reviewers = target_node['properties'].get('codelogic.reviewers', []) if target_node else []
-    
+
     # If target node doesn't have owners/reviewers, try to find them from the class or file node
     if not code_owners or not code_reviewers:
         class_node = None
         if class_name:
             class_node = next((n for n in nodes if n['primaryLabel'].endswith('ClassEntity') and class_name.lower() in n['name'].lower()), None)
-        
+
         if class_node:
             if not code_owners:
                 code_owners = class_node['properties'].get('codelogic.owners', [])
@@ -295,56 +295,8 @@ async def handle_method_impact(arguments: dict | None) -> list[types.TextContent
                         app_dependencies[app_name] = []
                     app_dependencies[app_name].append(depends_on)
 
-    # Identify REST endpoints or API controllers that might be affected
-    rest_endpoints = []
-    api_controllers = []
-    endpoint_nodes = []
-
-    # Look for Endpoint nodes directly
-    for node_item in nodes:
-        # Check for Endpoint primary label
-        if node_item.get('primaryLabel') == 'Endpoint':
-            endpoint_nodes.append({
-                'name': node_item.get('name', ''),
-                'path': node_item.get('properties', {}).get('path', ''),
-                'http_verb': node_item.get('properties', {}).get('httpVerb', ''),
-                'id': node_item.get('id')
-            })
-
-        # Check for controller types
-        if any(term in node_item.get('primaryLabel', '').lower() for term in
-                ['controller', 'restendpoint', 'apiendpoint', 'webservice']):
-            api_controllers.append({
-                'name': node_item.get('name', ''),
-                'type': node_item.get('primaryLabel', '')
-            })
-
-        # Check for REST annotations on methods
-        if node_item.get('primaryLabel') in ['JavaMethodEntity', 'DotNetMethodEntity']:
-            annotations = node_item.get('properties', {}).get('annotations', [])
-            if annotations and any(
-                    anno.lower() in str(annotations).lower() for anno in
-                    [
-                        'getmapping', 'postmapping', 'putmapping', 'deletemapping',
-                        'requestmapping', 'httpget', 'httppost', 'httpput', 'httpdelete'
-                    ]):
-                rest_endpoints.append({
-                    'name': node_item.get('name', ''),
-                    'annotation': str([a for a in annotations if any(m in a.lower() for m in ['mapping', 'http'])])
-                })
-
-    # Look for endpoint-to-endpoint dependencies
-    endpoint_dependencies = []
-    for rel in impact_data.get('data', {}).get('relationships', []):
-        if rel.get('type') in ['INVOKES_ENDPOINT', 'REFERENCES_ENDPOINT']:
-            start_node = find_node_by_id(impact_data.get('data', {}).get('nodes', []), rel.get('startId'))
-            end_node = find_node_by_id(impact_data.get('data', {}).get('nodes', []), rel.get('endId'))
-
-            if start_node and end_node:
-                endpoint_dependencies.append({
-                    'source': start_node.get('name', 'Unknown'),
-                    'target': end_node.get('name', 'Unknown')
-                })
+    # Use the new utility function to detect API endpoints and controllers
+    endpoint_nodes, rest_endpoints, api_controllers, endpoint_dependencies = find_api_endpoints(nodes, impact_data.get('data', {}).get('relationships', []))
 
     # Format nodes with metrics in markdown table format
     nodes_table = "| Name | Type | Complexity | Instruction Count | Method Count | Outgoing Refs | Incoming Refs |\n"
@@ -491,7 +443,7 @@ async def handle_method_impact(arguments: dict | None) -> list[types.TextContent
             impact_description += f"👤 **Code Owners**: Changes to this code should be reviewed by: {', '.join(code_owners)}\n"
         if code_reviewers:
             impact_description += f"👁️ **Preferred Reviewers**: Consider getting reviews from: {', '.join(code_reviewers)}\n"
-        
+
         if code_owners:
             impact_description += "\nConsult with the code owners before making significant changes to ensure alignment with original design intent.\n"
 

src/codelogic_mcp_server/server.py

@@ -17,6 +17,7 @@
 from mcp.server import NotificationOptions, Server
 import mcp.server.stdio
 from mcp.server.models import InitializationOptions
+from . import utils
 
 # Only load from .env file if we're not running tests
 # This allows tests to set their own environment variables
@@ -51,7 +52,7 @@ async def main():
                 write_stream,
                 InitializationOptions(
                     server_name="codelogic-mcp-server",
-                    server_version="0.1.0",
+                    server_version=utils.get_package_version(),
                     capabilities=server.get_capabilities(
                         notification_options=NotificationOptions(),
                         experimental_capabilities={},

src/codelogic_mcp_server/utils.py

@@ -16,10 +16,36 @@
 import sys
 import httpx
 import json
+import toml
 from datetime import datetime, timedelta
 from typing import Dict, Any, List
 import urllib.parse
 
+def get_package_version() -> str:
+    """
+    Get the package version from pyproject.toml.
+
+    Returns:
+        str: The package version from pyproject.toml
+
+    Raises:
+        FileNotFoundError: If pyproject.toml cannot be found
+        KeyError: If version cannot be found in pyproject.toml
+    """
+    try:
+        # Get the directory containing this file
+        current_dir = os.path.dirname(os.path.abspath(__file__))
+        # Go up to the project root (where pyproject.toml is)
+        project_root = os.path.dirname(os.path.dirname(current_dir))
+        pyproject_path = os.path.join(project_root, 'pyproject.toml')
+
+        with open(pyproject_path, 'r') as f:
+            config = toml.load(f)
+            return config['project']['version']
+    except Exception as e:
+        print(f"Warning: Could not read version from pyproject.toml: {e}", file=sys.stderr)
+        return "0.0.0"  # Fallback version if we can't read pyproject.toml
+
 # Cache TTL settings from environment variables (in seconds)
 TOKEN_CACHE_TTL = int(os.getenv('CODELOGIC_TOKEN_CACHE_TTL', '3600'))  # Default 1 hour
 METHOD_CACHE_TTL = int(os.getenv('CODELOGIC_METHOD_CACHE_TTL', '300'))  # Default 5 minutes
@@ -498,7 +524,7 @@ def process_database_entity_impact(impact_data, entity_type, entity_name, entity
     # we'll gather this information from the code entities that reference them
     code_owners = set()
     code_reviewers = set()
-    
+
     # Check code entities that reference this database entity
     for code_item in dependent_code:
         code_id = code_item.get("id")
@@ -508,7 +534,7 @@ def process_database_entity_impact(impact_data, entity_type, entity_name, entity
             reviewers = code_node.get('properties', {}).get('codelogic.reviewers', [])
             code_owners.update(owners)
             code_reviewers.update(reviewers)
-            
+
             # Look for parent classes that might contain ownership info
             for rel in impact_data.get('data', {}).get('relationships', []):
                 if rel.get('type').startswith('CONTAINS_') and rel.get('endId') == code_id:
@@ -711,6 +737,76 @@ def traverse_relationships(current_id):
     return applications
 
 
+def find_api_endpoints(nodes, relationships):
+    """
+    Find API endpoints, controllers, and their dependencies in impact data.
+
+    Args:
+        nodes (list): List of nodes from impact analysis
+        relationships (list): List of relationships from impact analysis
+
+    Returns:
+        tuple: (endpoint_nodes, rest_endpoints, api_controllers, endpoint_dependencies)
+            - endpoint_nodes: Explicit endpoint nodes
+            - rest_endpoints: Methods with REST annotations
+            - api_controllers: Controller classes
+            - endpoint_dependencies: Dependencies between endpoints
+    """
+    # Find explicit endpoints
+    endpoint_nodes = []
+    for node_item in nodes:
+        # Check for Endpoint primary label
+        if node_item.get('primaryLabel') == 'Endpoint':
+            endpoint_nodes.append({
+                'name': node_item.get('name', ''),
+                'path': node_item.get('properties', {}).get('path', ''),
+                'http_verb': node_item.get('properties', {}).get('httpVerb', ''),
+                'id': node_item.get('id')
+            })
+
+    # Find REST-annotated methods
+    rest_endpoints = []
+    api_controllers = []
+
+    for node_item in nodes:
+        # Check for controller types
+        if any(term in node_item.get('primaryLabel', '').lower() for term in
+               ['controller', 'restendpoint', 'apiendpoint', 'webservice']):
+            api_controllers.append({
+                'name': node_item.get('name', ''),
+                'type': node_item.get('primaryLabel', '')
+            })
+
+        # Check for REST annotations on methods
+        if node_item.get('primaryLabel') in ['JavaMethodEntity', 'DotNetMethodEntity']:
+            annotations = node_item.get('properties', {}).get('annotations', [])
+            if annotations and any(
+                    anno.lower() in str(annotations).lower() for anno in
+                    [
+                        'getmapping', 'postmapping', 'putmapping', 'deletemapping',
+                        'requestmapping', 'httpget', 'httppost', 'httpput', 'httpdelete'
+                    ]):
+                rest_endpoints.append({
+                    'name': node_item.get('name', ''),
+                    'annotation': str([a for a in annotations if any(m in a.lower() for m in ['mapping', 'http'])])
+                })
+
+    # Find endpoint dependencies
+    endpoint_dependencies = []
+    for rel in relationships:
+        if rel.get('type') in ['INVOKES_ENDPOINT', 'REFERENCES_ENDPOINT']:
+            start_node = find_node_by_id(nodes, rel.get('startId'))
+            end_node = find_node_by_id(nodes, rel.get('endId'))
+
+            if start_node and end_node:
+                endpoint_dependencies.append({
+                    'source': start_node.get('name', 'Unknown'),
+                    'target': end_node.get('name', 'Unknown')
+                })
+
+    return endpoint_nodes, rest_endpoints, api_controllers, endpoint_dependencies
+
+
 def generate_combined_database_report(entity_type, search_name, table_or_view, search_results, all_impacts):
     """
     Generate a combined report for all database entities.
@@ -753,7 +849,7 @@ def generate_combined_database_report(entity_type, search_name, table_or_view, s
         # Add code ownership information if available
         code_owners = impact.get("code_owners", [])
         code_reviewers = impact.get("code_reviewers", [])
-        
+
         if code_owners or code_reviewers:
             report += "#### Code Ownership\n"
             if code_owners: