Postgres MCP Pro Plus

Instructions

Implementation Reference

• src/postgres_mcp_pro_plus/server.py:686-714 (registration)

  MCP tool registration for 'analyze_schema_relationships'. Instantiates SchemaMappingTool with SQL driver, fetches user schemas, calls the tool's analyze_schema_relationships method, and formats the text response.
  @mcp.tool(description="Analyze schema relationships and dependencies with visual representation") async def analyze_schema_relationships() -> ResponseType: """Analyze inter-schema dependencies and relationships with visual representation data.""" try: sql_driver = await get_sql_driver() mapping_tool = SchemaMappingTool(sql_driver) # Get user schemas user_schemas_query = """ SELECT schema_name FROM information_schema.schemata WHERE schema_name NOT IN ('information_schema', 'pg_catalog', 'pg_toast') AND schema_name NOT LIKE 'pg_temp_%' AND schema_name NOT LIKE 'pg_toast_temp_%' ORDER BY schema_name """ rows = await sql_driver.execute_query(user_schemas_query) user_schemas = [row.cells["schema_name"] for row in rows] if rows else [] # Analyze schema relationships result = await mapping_tool.analyze_schema_relationships(user_schemas) return format_text_response(result) except Exception as e: logger.error(f"Error analyzing schema relationships: {e}") return format_error_response(str(e))
• src/postgres_mcp_pro_plus/schema_mapping.py:79-105 (handler)

  Core handler in SchemaMappingTool class that performs the schema relationship analysis: resets state, initializes SchemaNode instances, analyzes schemas and tables via helpers, builds mappings, generates results, and formats as text.
  async def analyze_schema_relationships(self, schemas: list[str]) -> str: """Analyze relationships between schemas and generate mapping data.""" try: # Reset state self.schema_nodes = {} self.table_nodes = {} self.cross_schema_relationships = [] self.intra_schema_relationships = [] # Initialize schema nodes for schema in schemas: self.schema_nodes[schema] = SchemaNode(name=schema) # Analyze each schema for schema in schemas: await self._analyze_schema(schema) # Build relationship mappings await self._build_relationship_mappings() # Generate analysis results result = await self._generate_analysis_results() return self._format_as_text(result) except Exception as e: logger.error(f"Error analyzing schema relationships: {e}") return f"Error analyzing schema relationships: {e}"
• src/postgres_mcp_pro_plus/schema_mapping.py:200-235 (helper)

  Helper method that aggregates analysis from schema dependencies, table dependencies, relationship patterns, visual data, and recommendations into a comprehensive result dictionary.
  async def _generate_analysis_results(self) -> dict[str, Any]: """Generate comprehensive analysis results.""" try: # Schema analysis schema_analysis = self._analyze_schema_dependencies() # Table analysis table_analysis = self._analyze_table_dependencies() # Relationship patterns relationship_patterns = self._analyze_relationship_patterns() # Visual representation data visual_data = self._generate_visual_representation() # Recommendations recommendations = self._generate_recommendations() return { "schema_analysis": schema_analysis, "table_analysis": table_analysis, "relationship_patterns": relationship_patterns, "visual_representation": visual_data, "recommendations": recommendations, "summary": { "total_schemas": len(self.schema_nodes), "total_tables": len(self.table_nodes), "cross_schema_relationships": len(self.cross_schema_relationships), "intra_schema_relationships": len(self.intra_schema_relationships), }, } except Exception as e: logger.error(f"Error generating analysis results: {e}") raise
• src/postgres_mcp_pro_plus/schema_mapping.py:18-39 (schema)

  Dataclass defining SchemaNode with fields and properties for dependency and isolation scoring, used throughout the analysis.
  @dataclass class SchemaNode: """Represents a schema node in the dependency graph.""" name: str table_count: int = 0 total_size_bytes: int = 0 total_rows: int = 0 outgoing_references: set[str] = field(default_factory=set) incoming_references: set[str] = field(default_factory=set) self_references: int = 0 @property def dependency_score(self) -> float: """Calculate dependency score based on incoming and outgoing references.""" return len(self.incoming_references) * 2 + len(self.outgoing_references) @property def isolation_score(self) -> float: """Calculate isolation score (lower is more isolated).""" return len(self.incoming_references) + len(self.outgoing_references)
• src/postgres_mcp_pro_plus/schema_mapping.py:41-67 (schema)

  Dataclass defining TableNode with fields and properties for connection count, hub status, and isolation, used in table-level analysis.
  @dataclass class TableNode: """Represents a table node in the dependency graph.""" schema: str name: str qualified_name: str size_bytes: int = 0 row_count: int = 0 outgoing_fks: list[str] = field(default_factory=list) incoming_fks: list[str] = field(default_factory=list) @property def connection_count(self) -> int: """Total number of connections (incoming + outgoing).""" return len(self.outgoing_fks) + len(self.incoming_fks) @property def is_hub(self) -> bool: """Check if table is a hub (has many incoming references).""" return len(self.incoming_fks) >= 3 @property def is_isolated(self) -> bool: """Check if table has no foreign key relationships.""" return len(self.outgoing_fks) == 0 and len(self.incoming_fks) == 0