CodeGraph CLI MCP Server

#!/usr/bin/env python3 """ Test agentic tools via HTTP using official MCP Python SDK. Properly follows MCP protocol with streamable HTTP transport. """ import asyncio import json import os import sys from pathlib import Path from datetime import datetime # Load .env try: from dotenv import load_dotenv env_path = Path(__file__).resolve().parent / ".env" if env_path.exists(): load_dotenv(env_path) except ImportError: pass # Configuration HTTP_HOST = os.environ.get("CODEGRAPH_HTTP_HOST", "127.0.0.1") HTTP_PORT = os.environ.get("CODEGRAPH_HTTP_PORT", "3003") SERVER_URL = f"http://{HTTP_HOST}:{HTTP_PORT}/mcp" # Test cases for consolidated 4 agentic tools # Each tuple: (tool_name, query, focus (optional), timeout) AGENTIC_TESTS = [ # agentic_context tests (absorbs: code_search, context_builder, semantic_question) ("agentic_context", "How is configuration loaded in this codebase? Find all config loading mechanisms.", None, 300), ("agentic_context", "Gather comprehensive context about the tier-aware prompt selection system", "builder", 300), ("agentic_context", "How does the LRU cache work in GraphToolExecutor? What gets cached and when?", "question", 300), # agentic_impact tests (absorbs: dependency_analysis, call_chain_analysis) ("agentic_impact", "Analyze the dependency chain for the PromptSelector. What does it depend on?", "dependencies", 300), ("agentic_impact", "Trace the call chain from execute_agentic_workflow to the graph analysis tools", "call_chain", 300), # agentic_architecture tests (absorbs: architecture_analysis, api_surface_analysis) ("agentic_architecture", "Analyze the architecture of the MCP server. Find coupling metrics and hub nodes.", "structure", 300), ("agentic_architecture", "What is the public API surface of the GraphToolExecutor?", "api_surface", 300), # agentic_quality tests (absorbs: complexity_analysis) ("agentic_quality", "Find the highest complexity hotspots in the codebase. Which functions have the highest risk scores?", None, 300), ] async def run_tests(): from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client print("\n" + "=" * 72) print("CodeGraph HTTP Agentic Tools Test (Official MCP SDK)") print("Testing 4 Consolidated Agentic Tools") print("=" * 72) print(f"Server: {SERVER_URL}") print("=" * 72 + "\n") try: async with streamablehttp_client(SERVER_URL) as (read_stream, write_stream, _): async with ClientSession(read_stream, write_stream) as session: # Initialize print("Initializing MCP connection...") await session.initialize() print("✓ MCP connection initialized\n") results = [] # Create output directory os.makedirs("test_output_http", exist_ok=True) for i, test_case in enumerate(AGENTIC_TESTS, 1): tool_name, query, focus, timeout = test_case print(f"{'=' * 72}") print(f"Testing: {tool_name}" + (f" (focus={focus})" if focus else "")) print(f"Query: {query}") print(f"Timeout: {timeout}s") print('=' * 72) start_time = asyncio.get_event_loop().time() result_text = None structured_output = None file_locations = [] try: # Build parameters params = {"query": query} if focus: params["focus"] = focus # Call tool with timeout result = await asyncio.wait_for( session.call_tool(tool_name, params), timeout=timeout ) # Parse result if result and len(result.content) > 0: result_text = result.content[0].text data = json.loads(result_text) # Prefer structured_output if provided if "structured_output" in data: structured_output = data["structured_output"] # Fallback: parse answer if it looks like JSON elif "answer" in data and isinstance(data["answer"], str): try: structured_output = json.loads(data["answer"]) except json.JSONDecodeError: structured_output = None # Extract file locations from structured output if structured_output: for field in ['components', 'hub_nodes', 'evidence', 'core_components', 'items', 'highlights']: if field in structured_output: for item in structured_output[field]: if isinstance(item, dict) and 'file_path' in item: file_locations.append(item) duration = asyncio.get_event_loop().time() - start_time steps = data.get("steps_taken", "?") if result_text else "?" print(f"\nℹ️ Completed in {duration:.1f}s ({steps} steps)") if structured_output: print(f" 📊 Structured Output: PRESENT") if file_locations: print(f" 📁 File Locations: {len(file_locations)}") for loc in file_locations[:3]: line = f":{loc['line_number']}" if loc.get('line_number') else "" print(f" - {loc.get('name', loc.get('title', 'unnamed'))} in {loc['file_path']}{line}") if len(file_locations) > 3: print(f" ... and {len(file_locations) - 3} more") else: print(" 📊 Structured Output: (none parsed)") results.append({ "test": tool_name, "focus": focus, "files": len(file_locations), "duration": duration, }) except asyncio.TimeoutError: duration = asyncio.get_event_loop().time() - start_time print(f"\n❌ TIMEOUT after {duration:.1f}s") results.append({"test": tool_name, "focus": focus, "files": 0, "duration": duration}) except Exception as e: duration = asyncio.get_event_loop().time() - start_time print(f"\n❌ ERROR: {e}") results.append({"test": tool_name, "focus": focus, "files": 0, "duration": duration}) # Write log file try: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") focus_suffix = f"_{focus}" if focus else "" log_filename = f"test_output_http/{str(i).zfill(2)}_{tool_name}{focus_suffix}_{timestamp}.log" with open(log_filename, "w") as f: f.write("=" * 80 + "\n") f.write(f"Test: {tool_name}\n") if focus: f.write(f"Focus: {focus}\n") f.write(f"Transport: HTTP (MCP SDK)\n") f.write(f"Timestamp: {timestamp}\n") f.write(f"Timeout: {timeout}s\n") f.write("=" * 80 + "\n\n") f.write("INPUT QUERY:\n") f.write("-" * 80 + "\n") f.write(f"{query}\n") f.write("-" * 80 + "\n\n") f.write("OUTPUT:\n") f.write("-" * 80 + "\n") if structured_output: f.write(json.dumps(structured_output, indent=2)) f.write("\n\n") f.write("FILE LOCATIONS EXTRACTED:\n") f.write("-" * 80 + "\n") for loc in file_locations: line_info = f":{loc['line_number']}" if loc.get('line_number') else "" f.write(f" {loc.get('name', loc.get('title', 'unnamed'))} in {loc['file_path']}{line_info}\n") elif result_text: f.write(result_text) else: f.write("(No result received)\n") f.write("-" * 80 + "\n\n") f.write(f"Duration: {duration:.1f}s\n") f.write("Status: RECORDED\n") print(f" 💾 Log saved: {log_filename}") except Exception as e: print(f" ⚠️ Failed to write log: {e}") # Summary print("\n" + "=" * 72) print("Test Summary") print("=" * 72) total_files = sum(r["files"] for r in results) total_time = sum(r["duration"] for r in results) print(f"Tests run: {len(results)} | File locations found: {total_files} | Total time: {total_time:.1f}s") print("=" * 72) return 0 except Exception as e: print(f"\n❌ Failed to connect to server: {e}") print(f"\nMake sure server is running:") print(f" codegraph start http --port {HTTP_PORT}") return 1 if __name__ == "__main__": sys.exit(asyncio.run(run_tests()))