"""
MCP Prompts Service
Provides intelligent PostgreSQL-focused prompts for database analysis, troubleshooting, and optimization
"""
import json
import logging
from typing import Dict, Any, List, Optional
from pathlib import Path
logger = logging.getLogger(__name__)
class MCPPromptsService:
"""Service that provides PostgreSQL-focused MCP Prompts"""
def __init__(self, config: Dict[str, Any]):
self.config = config
self._load_prompt_definitions()
def _load_prompt_definitions(self):
"""Load MCP prompt definitions from configuration"""
try:
prompts_file = Path(__file__).parent.parent / "resources" / "lists" / "mcp_prompts.json"
with open(prompts_file, 'r') as f:
definitions = json.load(f)
self.prompt_definitions = definitions
except Exception as e:
logger.warning(f"Could not load prompt definitions: {e}")
self.prompt_definitions = {}
def list_prompts(self) -> List[Dict[str, Any]]:
"""List all available MCP prompts"""
prompts = []
for category, prompt_list in self.prompt_definitions.items():
for prompt in prompt_list:
prompts.append({
"name": prompt["name"],
"description": prompt["description"],
"arguments": prompt.get("arguments", []),
"category": category
})
return prompts
async def get_prompt(self, name: str, arguments: Dict[str, Any] = None) -> Dict[str, Any]:
"""Get a specific prompt with populated arguments"""
if arguments is None:
arguments = {}
# Find the prompt definition
prompt_def = None
for category, prompt_list in self.prompt_definitions.items():
for prompt in prompt_list:
if prompt["name"] == name:
prompt_def = prompt
break
if prompt_def:
break
if not prompt_def:
raise ValueError(f"Prompt '{name}' not found")
# Build the contextualized prompt based on the prompt type
return await self._build_prompt_content(prompt_def, arguments)
async def _build_prompt_content(self, prompt_def: Dict[str, Any], arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Build the actual prompt content with context and instructions"""
prompt_content = self._get_base_prompt_content(prompt_def, arguments)
return {
"name": prompt_def["name"],
"description": prompt_def["description"],
"messages": [
{
"role": "user",
"content": {
"type": "text",
"text": prompt_content
}
}
]
}
def _get_base_prompt_content(self, prompt_def: Dict[str, Any], arguments: Dict[str, Any]) -> str:
"""Generate the base prompt content based on prompt type"""
name = prompt_def["name"]
args = arguments
if name == "analyze_database_schema":
database = args.get("database", "db3")
focus_area = args.get("focus_area", "overall")
return f"""You are a PostgreSQL database expert. Please analyze the database schema for {database} and provide comprehensive insights.
Focus area: {focus_area}
Please use the available MCP tools and resources to:
1. **Schema Overview**: Get the complete database schema using the database://schema resource
2. **Table Analysis**: Examine individual tables for structure and relationships
3. **Performance Context**: Check database statistics using database://stats resource
4. **Configuration Review**: Review PostgreSQL configuration via database://pgconfig resource
Based on your analysis, provide:
- **Schema Quality Assessment**: Evaluate normalization, data types, and design patterns
- **Relationship Mapping**: Identify and explain key relationships between tables
- **Performance Implications**: Highlight potential performance bottlenecks
- **Improvement Recommendations**: Suggest specific improvements for {focus_area}
- **Best Practices**: Note adherence to or deviations from PostgreSQL best practices
Format your response with clear sections and actionable insights."""
elif name == "diagnose_performance_issues":
database = args.get("database", "db3")
severity = args.get("severity_threshold", "medium")
return f"""You are a PostgreSQL performance specialist. Please diagnose potential performance issues in {database}.
Severity threshold: {severity} and above
Please use the available MCP tools and resources to:
1. **System Overview**: Check database statistics and configuration
2. **Log Analysis**: Examine PostgreSQL logs for errors and slow queries using get_postgresql_logs tool
3. **Query Performance**: Look for slow queries and performance patterns
4. **Resource Utilization**: Analyze table sizes, row counts, and storage usage
Provide a comprehensive performance diagnosis including:
- **Current Performance Metrics**: Key statistics and indicators
- **Identified Issues**: Specific problems found, categorized by severity
- **Root Cause Analysis**: Underlying causes of performance problems
- **Optimization Recommendations**: Specific, prioritized action items
- **Monitoring Suggestions**: Ongoing monitoring recommendations
Focus on issues that meet or exceed the {severity} severity threshold."""
elif name == "explain_table_relationships":
table_name = args.get("table_name", "")
include_semantic = args.get("include_semantic_context", "true").lower() == "true"
return f"""You are a database architect. Please explain the relationships and business context for the '{table_name}' table.
Include semantic context: {include_semantic}
Please use the available MCP tools and resources to:
1. **Table Structure**: Get detailed schema for {table_name} using table://{table_name}/schema
2. **Related Tables**: Identify connected tables through foreign keys and relationships
3. **Business Context**: Use find_related_data and semantic discovery tools
4. **Data Patterns**: Analyze data patterns if relevant
Provide a comprehensive explanation including:
- **Table Purpose**: What this table represents in the business domain
- **Column Analysis**: Meaning and purpose of each column
- **Relationship Mapping**: How this table connects to others (foreign keys, logical relationships)
- **Business Rules**: Implied business logic and constraints
- **Usage Patterns**: Common ways this table is likely queried or updated
{'- **Semantic Context**: Business meaning and domain context' if include_semantic else ''}
Make the explanation accessible to both technical and business stakeholders."""
elif name == "troubleshoot_connection_issues":
database = args.get("database", "db3")
time_range = args.get("time_range", "last_day")
return f"""You are a PostgreSQL systems administrator. Please troubleshoot connection issues for {database}.
Time range for analysis: {time_range}
Please use the available MCP tools and resources to:
1. **Configuration Analysis**: Review PostgreSQL configuration for connection settings
2. **Log Examination**: Analyze recent logs for connection errors and patterns
3. **Connection Statistics**: Check active connections and connection-related metrics
4. **System Health**: Overall database health indicators
Provide a systematic troubleshooting analysis:
- **Current Status**: Connection statistics and active sessions
- **Error Patterns**: Specific connection errors found in logs
- **Configuration Issues**: Connection-related configuration problems
- **Resource Constraints**: Memory, connection limits, or resource issues
- **Recommended Actions**: Step-by-step troubleshooting actions
- **Prevention Measures**: How to prevent similar issues
Focus on actionable solutions for the {time_range} time period."""
elif name == "investigate_slow_queries":
database = args.get("database", "db3")
min_duration = args.get("min_duration_ms", "1000")
return f"""You are a PostgreSQL query optimization expert. Please investigate slow queries in {database}.
Minimum duration threshold: {min_duration}ms
Please use the available MCP tools and resources to:
1. **Query Log Analysis**: Use get_postgresql_logs to find slow queries
2. **Performance Analysis**: Analyze query performance characteristics
3. **Schema Context**: Understand table structures involved in slow queries
4. **Optimization Opportunities**: Identify specific optimization potential
Provide a detailed slow query analysis:
- **Slow Query Identification**: Queries exceeding {min_duration}ms
- **Performance Metrics**: Execution times, frequency, resource usage
- **Query Analysis**: Structure and complexity assessment
- **Optimization Recommendations**: Specific improvements (indexes, rewrites, etc.)
- **Impact Assessment**: Expected performance gains from optimizations
- **Implementation Priority**: Order of recommended optimizations
Include specific SQL optimization suggestions where applicable."""
elif name == "check_database_health":
database = args.get("database", "db3")
include_recommendations = args.get("include_recommendations", "true").lower() == "true"
return f"""You are a PostgreSQL database administrator. Please perform a comprehensive health check of {database}.
Include recommendations: {include_recommendations}
Please use the available MCP tools and resources to:
1. **System Metrics**: Database statistics, sizes, and performance indicators
2. **Configuration Review**: PostgreSQL configuration analysis
3. **Log Analysis**: Recent logs for errors, warnings, and patterns
4. **Resource Usage**: Table sizes, connection counts, and resource utilization
Provide a comprehensive health assessment:
- **Overall Health Score**: Summary assessment (Excellent/Good/Fair/Poor)
- **Key Metrics**: Critical database health indicators
- **Issues Found**: Any problems or concerns identified
- **Performance Status**: Current performance characteristics
- **Security Assessment**: Basic security posture
{'- **Optimization Recommendations**: Specific improvement suggestions' if include_recommendations else ''}
{'- **Maintenance Schedule**: Recommended maintenance tasks' if include_recommendations else ''}
Present findings in order of priority with clear action items."""
elif name == "discover_data_patterns":
table_name = args.get("table_name", "")
analysis_type = args.get("analysis_type", "statistical")
sample_size = args.get("sample_size", "1000")
return f"""You are a data analyst specializing in PostgreSQL databases. Please discover and analyze data patterns in the '{table_name}' table.
Analysis type: {analysis_type}
Sample size: {sample_size} rows
Please use the available MCP tools and resources to:
1. **Table Structure**: Understand the table schema and data types
2. **Data Sampling**: Use analyze_data_patterns tool for pattern discovery
3. **Statistical Analysis**: Generate statistical summaries and distributions
4. **Anomaly Detection**: Identify unusual patterns or outliers
Provide detailed pattern analysis:
- **Data Overview**: Table structure and basic statistics
- **Pattern Discovery**: Key patterns found in the data
- **Statistical Summary**: Distributions, ranges, and statistical measures
- **Data Quality**: Completeness, consistency, and quality indicators
- **Anomalies**: Unusual patterns or potential data quality issues
- **Business Insights**: What the patterns reveal about the business domain
- **Recommendations**: Data management and quality improvement suggestions
Focus on {analysis_type} analysis with insights from {sample_size} sample rows."""
elif name == "find_business_insights":
business_question = args.get("business_question", "")
database = args.get("database", "db3")
include_viz = args.get("include_visualizations", "false").lower() == "true"
return f"""You are a business intelligence analyst with PostgreSQL expertise. Please investigate: "{business_question}"
Database: {database}
Include visualization suggestions: {include_viz}
Please use the available MCP tools and resources to:
1. **Concept Search**: Use search_tables_for_concept to find relevant data
2. **Relationship Discovery**: Use find_related_data to understand connections
3. **Data Analysis**: Examine relevant tables and their relationships
4. **Business Context**: Apply semantic understanding to data relationships
Provide comprehensive business insights:
- **Relevant Data Sources**: Tables and columns related to the business question
- **Data Relationships**: How different data elements connect
- **Key Findings**: Direct insights answering the business question
- **Supporting Evidence**: Data patterns and statistics that support findings
- **Business Implications**: What the insights mean for business decisions
{'- **Visualization Recommendations**: Suggested charts and visualizations' if include_viz else ''}
- **Further Investigation**: Additional questions or analysis suggested
Make insights actionable and business-focused."""
elif name == "generate_data_summary":
table_names = args.get("table_names", "")
include_sample = args.get("include_sample_data", "false").lower() == "true"
tables = [t.strip() for t in table_names.split(",") if t.strip()]
return f"""You are a database documentation specialist. Please create a comprehensive summary of the following tables: {', '.join(tables)}
Include sample data: {include_sample}
Please use the available MCP tools and resources to:
1. **Schema Analysis**: Get detailed structure for each table
2. **Relationship Mapping**: Understand how tables relate to each other
3. **Statistics Gathering**: Row counts, sizes, and performance metrics
4. **Semantic Context**: Business meaning and domain context
Provide a complete data summary:
- **Executive Summary**: High-level overview of the table group
- **Individual Table Summaries**: Purpose, structure, and key characteristics
- **Data Relationships**: How tables connect and relate to each other
- **Business Context**: Domain meaning and business rules
- **Data Statistics**: Volumes, growth patterns, and usage metrics
{'- **Sample Data**: Representative examples from each table' if include_sample else ''}
- **Usage Patterns**: Common access patterns and query types
- **Recommendations**: Data management and optimization suggestions
Format as comprehensive documentation suitable for stakeholders."""
elif name == "suggest_query_optimization":
sql_query = args.get("sql_query", "")
database = args.get("database", "db3")
goal = args.get("optimization_goal", "speed")
return f"""You are a PostgreSQL query optimization expert. Please optimize this SQL query:
```sql
{sql_query}
```
Database context: {database}
Optimization goal: {goal}
Please use the available MCP tools and resources to:
1. **Query Analysis**: Understand the query structure and complexity
2. **Schema Context**: Analyze tables and columns involved
3. **Performance Testing**: Use analyze_query_performance tool
4. **Index Assessment**: Review current indexing strategy
Provide detailed optimization recommendations:
- **Current Query Analysis**: Structure, complexity, and potential issues
- **Performance Assessment**: Current execution characteristics
- **Optimization Strategies**: Specific techniques for improving {goal}
- **Rewritten Query**: Optimized version of the original query
- **Index Recommendations**: Suggested indexes to support the query
- **Execution Plan**: Expected improvements in query plan
- **Trade-offs**: Any compromises made for the optimization goal
- **Testing Approach**: How to validate the optimization improvements
Include the optimized SQL code with clear explanations."""
elif name == "recommend_index_strategy":
table_name = args.get("table_name", "")
query_patterns = args.get("query_patterns", "")
return f"""You are a PostgreSQL indexing specialist. Please recommend an optimal indexing strategy for the '{table_name}' table.
Query patterns: {query_patterns}
Please use the available MCP tools and resources to:
1. **Table Analysis**: Understand table structure, size, and current indexes
2. **Usage Patterns**: Analyze how the table is typically queried
3. **Performance Impact**: Assess current query performance
4. **Index Opportunities**: Identify optimal indexing strategies
Provide comprehensive indexing recommendations:
- **Current Index Analysis**: Existing indexes and their effectiveness
- **Query Pattern Assessment**: How queries interact with the table
- **Recommended Indexes**: Specific indexes to create or modify
- **Index Types**: B-tree, GIN, GiST, or other specialized indexes
- **Performance Impact**: Expected query performance improvements
- **Maintenance Considerations**: Storage and maintenance overhead
- **Implementation Order**: Prioritized order for implementing indexes
- **Monitoring Strategy**: How to track index effectiveness
Include specific CREATE INDEX statements where appropriate."""
elif name == "analyze_storage_optimization":
database = args.get("database", "db3")
focus_tables = args.get("focus_tables", "")
focus_list = [t.strip() for t in focus_tables.split(",") if t.strip()] if focus_tables else []
return f"""You are a PostgreSQL storage optimization expert. Please analyze storage usage and recommend optimizations for {database}.
{"Focus tables: " + ", ".join(focus_list) if focus_list else "Analyzing all tables"}
Please use the available MCP tools and resources to:
1. **Storage Analysis**: Use table size information and database statistics
2. **Usage Patterns**: Understand data access and modification patterns
3. **Optimization Opportunities**: Identify storage efficiency improvements
4. **Performance Impact**: Balance storage optimization with query performance
Provide detailed storage optimization analysis:
- **Storage Overview**: Current storage usage and distribution
- **Large Tables**: Tables consuming significant storage space
- **Growth Patterns**: Historical growth and projections
- **Optimization Opportunities**: Specific storage reduction strategies
- **Compression Options**: Row-level and table-level compression possibilities
- **Archival Strategies**: Data archival and purging recommendations
- **Performance Trade-offs**: Impact of storage optimizations on query performance
- **Implementation Plan**: Step-by-step optimization implementation
{"Focus specifically on: " + ", ".join(focus_list) if focus_list else "Cover all significant tables in the analysis"}."""
elif name == "security_audit_database":
database = args.get("database", "db3")
audit_scope = args.get("audit_scope", "configuration")
return f"""You are a PostgreSQL security specialist. Please perform a security audit of {database}.
Audit scope: {audit_scope}
Please use the available MCP tools and resources to:
1. **Configuration Security**: Review PostgreSQL configuration for security settings
2. **Access Analysis**: Examine connection patterns and access controls
3. **Log Review**: Check logs for security-related events
4. **Vulnerability Assessment**: Identify potential security vulnerabilities
Provide a comprehensive security audit:
- **Security Posture**: Overall security assessment and score
- **Configuration Review**: Security-relevant configuration settings
- **Access Control Analysis**: User permissions and connection security
- **Vulnerability Identification**: Potential security risks found
- **Compliance Assessment**: Adherence to security best practices
- **Threat Analysis**: Potential attack vectors and risks
- **Remediation Plan**: Prioritized security improvements
- **Monitoring Recommendations**: Ongoing security monitoring suggestions
Focus on {audit_scope} with actionable security recommendations."""
elif name == "analyze_access_patterns":
database = args.get("database", "db3")
time_window = args.get("time_window", "last_day")
return f"""You are a PostgreSQL security analyst. Please analyze database access patterns for {database}.
Time window: {time_window}
Please use the available MCP tools and resources to:
1. **Connection Analysis**: Review connection logs and patterns
2. **Query Pattern Review**: Analyze query types and frequencies
3. **User Activity**: Examine user access patterns and behaviors
4. **Anomaly Detection**: Identify unusual or suspicious activities
Provide detailed access pattern analysis:
- **Access Summary**: Overall access statistics for {time_window}
- **Connection Patterns**: Normal vs. unusual connection behaviors
- **Query Analysis**: Types and patterns of queries executed
- **User Behavior**: Individual user access patterns and activities
- **Anomaly Report**: Suspicious or unusual activities detected
- **Risk Assessment**: Security implications of observed patterns
- **Baseline Establishment**: Normal patterns for future comparison
- **Monitoring Recommendations**: Improved access monitoring strategies
Focus on security-relevant patterns and potential threats."""
else:
return f"""Please help with the task: {name}
Use the available MCP tools and resources to gather relevant information and provide a comprehensive analysis.
Arguments provided: {json.dumps(arguments, indent=2)}
Provide detailed insights and actionable recommendations based on your analysis."""
def get_prompt_info(self) -> Dict[str, Any]:
"""Get information about available prompts"""
categories = {}
total_prompts = 0
for category, prompt_list in self.prompt_definitions.items():
categories[category] = {
"count": len(prompt_list),
"prompts": [p["name"] for p in prompt_list]
}
total_prompts += len(prompt_list)
return {
"total_prompts": total_prompts,
"categories": categories,
"prompt_capabilities": [
"Database schema analysis and optimization",
"Performance troubleshooting and tuning",
"Data pattern discovery and insights",
"Security auditing and access analysis",
"Query optimization and indexing strategies",
"Storage optimization and maintenance planning"
]
}
