postgres-mcp

Instructions

Args:
    search_term: Column name pattern to search (case-insensitive)
    schema: Optional schema to limit search (default: all user schemas)
    
Returns:
    List of matching columns with table information

Implementation Reference

• postgres_mcp/server.py:394-415 (handler)

  MCP tool handler for 'search_columns', decorated with @mcp.tool(). Delegates to PostgresClient.search_columns and formats the response.

  @mcp.tool()
  @handle_db_error
  def search_columns(search_term: str, schema: Optional[str] = None) -> dict:
      """Search for columns by name across all tables.
      
      Args:
          search_term: Column name pattern to search (case-insensitive)
          schema: Optional schema to limit search (default: all user schemas)
          
      Returns:
          List of matching columns with table information
      """
      client = get_client()
      columns = client.search_columns(search_term, schema)
      
      return {
          "search_term": search_term,
          "schema_filter": schema,
          "matches": columns,
          "count": len(columns),
      }

• postgres_mcp/postgres_client.py:540-578 (helper)

  Core implementation of column search logic in PostgresClient class, executing SQL query against information_schema.columns.

  def search_columns(self, search_term: str, schema: Optional[str] = None) -> list[dict]:
      """Search for columns by name.
      
      Args:
          search_term: Column name pattern (supports LIKE wildcards)
          schema: Optional schema filter
          
      Returns:
          List of matching columns with table info
      """
      # Sanitize search term for LIKE pattern
      search_pattern = f"%{search_term}%"
      
      query = """
          SELECT 
              table_schema,
              table_name,
              column_name,
              data_type,
              is_nullable
          FROM information_schema.columns
          WHERE column_name ILIKE %s
      """
      params = [search_pattern]
      
      if schema:
          query += " AND table_schema = %s"
          params.append(schema)
      
      query += """
          AND table_schema NOT IN ('information_schema', 'pg_catalog')
          ORDER BY table_schema, table_name, column_name
          LIMIT 100
      """
      
      with self.get_cursor() as cursor:
          cursor.execute(query, params)
          return [dict(row) for row in cursor.fetchall()]