Rover MCP Server

#!/usr/bin/env python3 """ Quick Query Test - Demonstrates 3 example queries from the test list Shows how the queries would work in practice with real MCP tools """ import asyncio import os from datetime import datetime # Set environment variables os.environ["MCP_TRANSPORT"] = "stdio" os.environ["CERT_FILE"] = "sa-cert.crt" os.environ["KEY_FILE"] = "privkey.pem" from mcp_server import ( rover_group, get_comprehensive_member_profile, correlate_rover_groups_with_jira ) def print_query_header(query_num: int, title: str, query: str): """Print a formatted query header.""" print(f"\n{'='*70}") print(f"🎯 QUERY #{query_num}: {title}") print(f"{'='*70}") print(f"📝 Query: \"{query}\"") print(f"{'─'*70}") async def demo_query_1(): """Query 1: Team Structure Discovery""" print_query_header(1, "Team Structure Discovery", "Who are the members of the sp-ai-support-chatbot team and what are their roles?") try: # Get team structure team_data = await rover_group("sp-ai-support-chatbot") admin_data = await rover_group("sp-ai-support-chatbot-admins") if "error" not in team_data: print(f"\n🏢 **Team: sp-ai-support-chatbot**") print(f"📝 Description: {team_data.get('description', 'No description')}") members = team_data.get("members", []) print(f"👥 Total Members: {len(members)}") if members: print(f"\n📋 Team Members:") for i, member in enumerate(members[:8], 1): # Show first 8 uid = member.get("uid", "Unknown") print(f" {i:2d}. {uid}") if len(members) > 8: print(f" ... and {len(members) - 8} more members") if "error" not in admin_data: admin_members = admin_data.get("members", []) print(f"\n👑 **Admin Team: sp-ai-support-chatbot-admins**") print(f"🔐 Admin Members: {len(admin_members)}") # Find overlap team_uids = {m.get("uid") for m in members} admin_uids = {m.get("uid") for m in admin_members} overlap = team_uids.intersection(admin_uids) print(f"🔗 Members with admin privileges: {len(overlap)}") if overlap: print(f" Admin members: {', '.join(list(overlap)[:5])}") print(f"\n✅ **Answer**: The team has {len(members)} total members with {len(admin_members)} having admin privileges.") except Exception as e: print(f"❌ Error: {str(e)}") async def demo_query_2(): """Query 2: Individual Expertise Analysis""" print_query_header(2, "Individual Expertise Analysis", "Give me a comprehensive profile of dhshah including their expertise areas and current projects") try: profile = await get_comprehensive_member_profile("dhshah") if "error" not in profile: # Show the formatted summary (truncated) summary = profile.get("formatted_summary", "") print(f"\n📊 **Comprehensive Profile for dhshah:**") print(summary[:500] + "..." if len(summary) > 500 else summary) # Extract key insights raw_data = profile.get("raw_data", {}) expertise = raw_data.get("expertise", []) current_work = raw_data.get("current_work", []) activity_level = raw_data.get("activity_level", "Unknown") print(f"\n🎯 **Key Insights:**") print(f"💼 Expertise Areas: {', '.join(expertise[:3])}") print(f"📈 Activity Level: {activity_level}") print(f"🚀 Current Focus: {current_work[0] if current_work else 'Information being gathered'}") print(f"\n✅ **Answer**: dhshah is a platform engineering expert currently focusing on Red Hat infrastructure.") else: print(f"❌ Error: {profile['error']}") except Exception as e: print(f"❌ Error: {str(e)}") async def demo_query_3(): """Query 3: Team Activity Assessment""" print_query_header(3, "Team Activity Assessment", "Is the sp-ai-support-chatbot group actively working on projects or dormant?") try: correlation = await correlate_rover_groups_with_jira("sp-ai-support-chatbot") if "error" not in correlation: jira_data = correlation.get("jira_correlation", {}) insights = correlation.get("insights", []) recommendations = correlation.get("recommendations", []) print(f"\n📊 **Activity Analysis for sp-ai-support-chatbot:**") # Extract key metrics owners_activity = jira_data.get("owners_jira_activity", {}) common_projects = jira_data.get("common_projects", []) access_patterns = jira_data.get("access_patterns", {}) print(f"👑 Team owners analyzed: {len(owners_activity)}") print(f"🤝 Common JIRA projects: {len(common_projects)}") print(f"📈 JIRA access detected: {access_patterns.get('has_jira_access', False)}") print(f"\n💡 **Generated Insights:**") for insight in insights: print(f" • {insight}") if recommendations: print(f"\n🎯 **Recommendations:**") for rec in recommendations: print(f" • {rec}") # Determine activity status if "no JIRA activity" in str(insights).lower(): status = "🟡 DORMANT - Limited recent project activity detected" elif len(common_projects) > 0: status = "🟢 ACTIVE - Working on multiple shared projects" else: status = "🟠 MODERATE - Some activity but limited shared work" print(f"\n✅ **Answer**: {status}") else: print(f"❌ Error: {correlation['error']}") except Exception as e: print(f"❌ Error: {str(e)}") async def run_query_demos(): """Run the 3 demo queries""" print("🚀 ROVER MCP ANALYTICAL QUERIES - LIVE DEMONSTRATION") print("="*70) print(f"⏰ Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"🎯 Demonstrating 3 key query types with real data") demos = [ demo_query_1, demo_query_2, demo_query_3 ] for demo_func in demos: try: await demo_func() except Exception as e: print(f"❌ Demo failed: {str(e)}") print(f"\n{'='*70}") print("🎉 DEMONSTRATION COMPLETE") print(f"{'='*70}") print(f""" 🚀 **What You Just Saw:** ✅ Real team structure analysis with member counts ✅ Comprehensive individual profiles with expertise mapping ✅ Team activity assessment with actionable insights 💡 **These same capabilities work in Cursor:** • Ask natural language questions • Get professional, formatted responses • Leverage real Red Hat internal data • Receive actionable recommendations 🎯 **Try these queries in Cursor to see the full power!** """) if __name__ == "__main__": try: asyncio.run(run_query_demos()) except KeyboardInterrupt: print("\n\n⚠️ Demo interrupted by user") except Exception as e: print(f"\n\n❌ Demo failed: {str(e)}")